scrapy / scrapy
1
"""Some debugging functions for working with the Scrapy engine"""
2

3
# used in global tests code
4 7
from time import time  # noqa: F401
5

6

7 7
def get_engine_status(engine):
8
    """Return a report of the current engine status"""
9 7
    tests = [
10
        "time()-engine.start_time",
11
        "engine.has_capacity()",
12
        "len(engine.downloader.active)",
13
        "engine.scraper.is_idle()",
14
        "engine.spider.name",
15
        "engine.spider_is_idle(engine.spider)",
16
        "engine.slot.closing",
17
        "len(engine.slot.inprogress)",
18
        "len(engine.slot.scheduler.dqs or [])",
19
        "len(engine.slot.scheduler.mqs)",
20
        "len(engine.scraper.slot.queue)",
21
        "len(engine.scraper.slot.active)",
22
        "engine.scraper.slot.active_size",
23
        "engine.scraper.slot.itemproc_size",
24
        "engine.scraper.slot.needs_backout()",
25
    ]
26

27 7
    checks = []
28 7
    for test in tests:
29 7
        try:
30 7
            checks += [(test, eval(test))]
31 0
        except Exception as e:
32 0
            checks += [(test, f"{type(e).__name__} (exception)")]
33

34 7
    return checks
35

36

37 7
def format_engine_status(engine=None):
38 0
    checks = get_engine_status(engine)
39 0
    s = "Execution engine status\n\n"
40 7
    for test, result in checks:
41 0
        s += f"{test:<47} : {result}\n"
42 0
    s += "\n"
43

44 0
    return s
45

46

47 7
def print_engine_status(engine):
48 0
    print(format_engine_status(engine))

Read our documentation on viewing source code .

Loading