(self, service: beat.Service)
| 133 | self.app.finalize() |
| 134 | |
| 135 | def startup_info(self, service: beat.Service) -> str: |
| 136 | scheduler = service.get_scheduler(lazy=True) |
| 137 | return STARTUP_INFO_FMT.format( |
| 138 | conninfo=self.app.connection().as_uri(), |
| 139 | timestamp=datetime.now().replace(microsecond=0), |
| 140 | logfile=self.logfile or '[stderr]', |
| 141 | loglevel=LOG_LEVELS[self.loglevel], |
| 142 | loader=qualname(self.app.loader), |
| 143 | scheduler=qualname(scheduler), |
| 144 | scheduler_info=scheduler.info, |
| 145 | hmax_interval=humanize_seconds(scheduler.max_interval), |
| 146 | max_interval=scheduler.max_interval, |
| 147 | ) |
| 148 | |
| 149 | def set_process_title(self) -> None: |
| 150 | arg_start = 'manage' in sys.argv[0] and 2 or 1 |
no test coverage detected