(self, parsed_args, parsed_globals)
| 209 | } |
| 210 | |
| 211 | def _run_main(self, parsed_args, parsed_globals): |
| 212 | logs_client = self._session.create_client( |
| 213 | 'logs', |
| 214 | region_name=parsed_globals.region, |
| 215 | endpoint_url=parsed_globals.endpoint_url, |
| 216 | verify=parsed_globals.verify_ssl, |
| 217 | ) |
| 218 | logs_generator = self._get_log_events_generator( |
| 219 | logs_client, parsed_args.follow |
| 220 | ) |
| 221 | log_events = logs_generator.iter_log_events( |
| 222 | parsed_args.group_name, |
| 223 | start=parsed_args.since, |
| 224 | filter_pattern=parsed_args.filter_pattern, |
| 225 | log_stream_names=parsed_args.log_stream_names, |
| 226 | log_stream_name_prefix=parsed_args.log_stream_name_prefix, |
| 227 | ) |
| 228 | self._output_log_events(parsed_args, parsed_globals, log_events) |
| 229 | return 0 |
| 230 | |
| 231 | def _get_log_events_generator(self, logs_client, follow): |
| 232 | timestamp_utils = TimestampUtils() |
nothing calls this directly
no test coverage detected