(self)
| 376 | return celery_python, celery |
| 377 | |
| 378 | def get_pid_by_command(self) -> Optional[int]: |
| 379 | celery_env = self.get_celery_env() |
| 380 | if not celery_env[0] or not celery_env[1]: |
| 381 | return super().get_pid_by_command() |
| 382 | target = [] |
| 383 | cmd_list = list(celery_env) + self.split_command()[1:] |
| 384 | for p in psutil.process_iter(["cmdline", "pid"]): |
| 385 | try: |
| 386 | if cmd_list == p.cmdline(): |
| 387 | target.append(p) |
| 388 | except: |
| 389 | continue |
| 390 | |
| 391 | for p in target: |
| 392 | try: |
| 393 | if p.ppid() not in target: |
| 394 | return p.pid |
| 395 | except: |
| 396 | continue |
| 397 | return None |
| 398 | |
| 399 | |
| 400 | class ServiceManager: |
nothing calls this directly
no test coverage detected