(self)
| 364 | class CeleryService(PythonService): |
| 365 | |
| 366 | def get_celery_env(self) -> Tuple[str, str]: |
| 367 | celery = "{}/celery".format(self.env.python_path) |
| 368 | if not os.path.isfile(celery): |
| 369 | return '', '' |
| 370 | celery_data = public.readFile(celery) |
| 371 | if not isinstance(celery_data, str): |
| 372 | return '', '' |
| 373 | celery_python = celery_data.split("\n", 1)[0] |
| 374 | if celery_python.startswith("#!"): |
| 375 | celery_python = celery_python[2:].strip() |
| 376 | return celery_python, celery |
| 377 | |
| 378 | def get_pid_by_command(self) -> Optional[int]: |
| 379 | celery_env = self.get_celery_env() |
no test coverage detected