Return a report of the current engine status
(engine: ExecutionEngine)
| 11 | |
| 12 | |
| 13 | def get_engine_status(engine: ExecutionEngine) -> list[tuple[str, Any]]: |
| 14 | """Return a report of the current engine status""" |
| 15 | tests = [ |
| 16 | "time()-engine.start_time", |
| 17 | "len(engine.downloader.active)", |
| 18 | "engine.scraper.is_idle()", |
| 19 | "engine.spider.name", |
| 20 | "engine.spider_is_idle()", |
| 21 | "engine._slot.closing", |
| 22 | "len(engine._slot.inprogress)", |
| 23 | "len(engine._slot.scheduler.dqs or [])", |
| 24 | "len(engine._slot.scheduler.mqs)", |
| 25 | "len(engine.scraper.slot.queue)", |
| 26 | "len(engine.scraper.slot.active)", |
| 27 | "engine.scraper.slot.active_size", |
| 28 | "engine.scraper.slot.itemproc_size", |
| 29 | "engine.scraper.slot.needs_backout()", |
| 30 | ] |
| 31 | |
| 32 | checks: list[tuple[str, Any]] = [] |
| 33 | for test in tests: |
| 34 | try: |
| 35 | checks += [(test, eval(test))] # noqa: S307 |
| 36 | except Exception as e: |
| 37 | checks += [(test, f"{type(e).__name__} (exception)")] |
| 38 | |
| 39 | return checks |
| 40 | |
| 41 | |
| 42 | def format_engine_status(engine: ExecutionEngine) -> str: |
no outgoing calls