#!/usr/bin/env python3 """Find deadlock, hang and starvation risks in asyncio and threaded Python code. Static analysis only: files are parsed with `ast`, never imported or executed. The scanner builds a per-module lock-order graph from nested `with lock:` / `async with lock:` blocks and `.acquire()` calls (following one level of calls to functions and methods defined in the same module), then reports cycles, re-entrant acquisitions and the classic blocking mistakes listed below. DLK001 lock-order-cycle two code paths take the same locks in opposite order DLK002 reentrant-acquire a non-reentrant Lock is acquired again while held (directly or via a call one level deep) DLK003 thread-lock-in-async threading/multiprocessing lock taken inside `async def` DLK004 sleep-in-async time.sleep() inside `async def` freezes the event loop DLK005 blocking-wait-in-async .result() on an executor future, .join(), sync queue get/put, Event/Condition .wait() or futures.wait() inside `async def` DLK006 threadsafe-result-in-loop run_coroutine_threadsafe(...).result() inside `async def` DLK007 nested-event-loop asyncio.run / run_until_complete / run_forever inside `async def` DLK008 join-without-task-done Queue.join() in a module that never calls task_done() DLK009 acquire-without-finally Lock/Semaphore .acquire() not paired with release() in finally DLK010 condition-wait-no-loop Condition.wait() not inside a `while` predicate loop DLK011 wait-without-timeout sync queue get/join or Event/Condition wait with no timeout (low) DLK012 executor-self-wait a task running in an executor submits to the same executor and waits for the result (pool exhaustion deadlock) DLK013 thread-never-joined non-daemon Thread started and never joined (shutdown hang) DLK014 process-join-before-drain Process.join() before draining its multiprocessing Queue DLK015 bounded-put-no-consumer bounded queue filled in a scope that never reads it (low) DLK016 unresolved-future `await fut` on a create_future() result nothing ever sets DLK017 fork-with-threads os.fork / "fork" start method in a module that starts threads DLK018 lock-in-signal-handler signal handler takes a lock or logs (re-entrancy deadlock) Suppress a finding on its line with `# noqa` or `# noqa: DLK001,DLK009`. Usage: python hang_scan.py app/worker.py python hang_scan.py src/ tests/ --json cat agent_loop.py | python hang_scan.py - python hang_scan.py src/ --strict --ignore DLK011,DLK015 python hang_scan.py src/ --select DLK001,DLK002,lock-order-cycle python hang_scan.py --list-rules Exit codes: 0 no findings (or findings without --strict), 1 findings with --strict, 2 usage error, unreadable file or syntax error in a file named on the command line (files found by walking a directory are warned about and skipped). No dependencies (standard library only, Python 3.9+). """ import argparse import ast import json import os import re import sys # ---------------------------------------------------------------- rules RULES = { "DLK001": ("lock-order-cycle", "high", "Locks acquired in inconsistent order across code paths (classic AB/BA deadlock)", "Two tasks/threads each hold one lock and wait forever for the other.", "Pick one global order and always take locks in it, e.g. sorted by name:\n" " first, second = sorted((a, b), key=id)\n" " with first, second:\n ...\n" "or merge them into one lock, or never hold one lock while taking another."), "DLK002": ("reentrant-acquire", "high", "Non-reentrant lock acquired again by the code that already holds it", "threading.Lock and asyncio.Lock are not reentrant: the second acquire waits for itself forever.", "Split the helper into a locked public method and an unlocked `_helper_locked()` that\n" "assumes the caller holds the lock; call the unlocked one while holding it.\n" "(threading.RLock is reentrant; asyncio has no RLock by design.)"), "DLK003": ("thread-lock-in-async", "medium", "threading/multiprocessing lock used inside async def", "A contended threading lock blocks the whole event loop thread; if the holder is another\n" "task on the same loop it can never release it, so the loop deadlocks.", "Use asyncio.Lock for state shared between tasks:\n" " lock = asyncio.Lock()\n async with lock:\n ...\n" "For state shared with threads, take the threading lock in a thread:\n" " await asyncio.to_thread(update_shared_state) # 3.9+"), "DLK004": ("sleep-in-async", "high", "time.sleep() inside async def", "time.sleep blocks the event loop thread: every other task (heartbeats, streams, timeouts) stops.", "await asyncio.sleep(delay)"), "DLK005": ("blocking-wait-in-async", "high", "Blocking wait (future.result(), .join(), sync queue, Event.wait) inside async def", "The event loop thread blocks; if the thing being waited for needs the loop, it never finishes.", "Await an asyncio equivalent or move the wait off the loop:\n" " result = await asyncio.wrap_future(fut) # concurrent.futures.Future\n" " await asyncio.to_thread(worker.join, 30) # thread/process join\n" " item = await asyncio.to_thread(sync_q.get, timeout=5)"), "DLK006": ("threadsafe-result-in-loop", "high", "run_coroutine_threadsafe(...).result() called from inside async def", "If the target loop is the current loop, .result() blocks the very loop that must run the\n" "coroutine: a guaranteed deadlock. Even for another loop it freezes this one.", "Inside the loop just await the coroutine (or a task):\n result = await coro\n" "Across loops: result = await asyncio.wrap_future(asyncio.run_coroutine_threadsafe(coro, other))"), "DLK007": ("nested-event-loop", "high", "asyncio.run / run_until_complete / run_forever inside async def", "A loop is already running in this thread: asyncio.run raises RuntimeError, and\n" "run_until_complete raises 'This event loop is already running' (or deadlocks with nest_asyncio).", "Await the coroutine directly: result = await main_step()\n" "In Jupyter use top-level `await`; in sync code called from a loop, use\n" "asyncio.run_coroutine_threadsafe from a worker thread."), "DLK008": ("join-without-task-done", "high", "Queue.join() but the module never calls task_done()", "join() waits until unfinished_tasks reaches 0; without task_done() it never does.", "Call task_done() exactly once per get(), in finally:\n" " item = await queue.get()\n try:\n await handle(item)\n" " finally:\n queue.task_done()"), "DLK009": ("acquire-without-finally", "high", "acquire() without a matching release() in finally", "An exception (429, timeout, cancellation) between acquire and release leaks the permit;\n" "after N leaks every caller waits forever (rate limiter starvation).", "Use the context manager:\n async with sem:\n return await call_llm(prompt)\n" "or acquire immediately before try/finally:\n" " lock.acquire()\n try:\n ...\n finally:\n lock.release()"), "DLK010": ("condition-wait-no-loop", "medium", "Condition.wait() not inside a while-predicate loop", "Wakeups can be spurious, stolen by another consumer, or arrive before wait() starts\n" "(lost wakeup): re-check the predicate in a loop.", "with cond:\n while not ready():\n cond.wait(timeout=1.0)\n" "or simply: cond.wait_for(ready, timeout=30)"), "DLK011": ("wait-without-timeout", "low", "Blocking queue get/join or Event/Condition wait without a timeout", "If the producer/notifier dies, this thread blocks forever and shutdown hangs.", "Pass a timeout and handle it:\n try:\n item = q.get(timeout=1.0)\n" " except queue.Empty:\n if stop.is_set():\n return\n" "or use Queue.shutdown() (3.13+) to wake blocked getters."), "DLK012": ("executor-self-wait", "high", "Executor task submits to its own executor and waits for the result", "Once all workers run outer tasks that wait on inner tasks, no worker is free to run\n" "the inner ones: thread pool exhaustion deadlock.", "Use a separate executor for inner work, run the inner work inline, or restructure so\n" "the caller fans out all leaf tasks itself (no waiting inside workers)."), "DLK013": ("thread-never-joined", "medium", "Non-daemon Thread started and never joined", "Interpreter shutdown waits for every non-daemon thread; a worker blocked on a queue\n" "makes the program hang at exit (Ctrl+C appears to do nothing).", "Keep a handle, signal it to stop, and join with a timeout:\n" " stop.set(); worker.join(timeout=5)\n" "or mark truly disposable helpers daemon=True."), "DLK014": ("process-join-before-drain", "high", "Process.join() before the multiprocessing Queue it writes to is drained", "A child that put() large items does not exit until its feeder thread flushes the pipe;\n" "the parent waits in join() and never reads: deadlock (documented in multiprocessing).", "Drain first, then join:\n results = [q.get(timeout=60) for _ in procs]\n" " for p in procs:\n p.join(timeout=10)"), "DLK015": ("bounded-put-no-consumer", "low", "Bounded queue is filled in a scope that never reads it and never hands it to a consumer", "After maxsize items put() blocks forever.", "Start the consumer before producing (and pass it the queue), or use an unbounded queue:\n" " async with asyncio.TaskGroup() as tg:\n tg.create_task(consume(q))\n" " for doc in docs:\n await q.put(doc)"), "DLK016": ("unresolved-future", "high", "await on a Future created with create_future() that nothing in the module resolves", "Nobody calls set_result/set_exception/cancel on it, so the awaiting task waits forever.", "Resolve it on every path (including errors) and bound the wait:\n" " fut = loop.create_future()\n pending[req_id] = fut # resolver: fut.set_result(...)\n" " return await asyncio.wait_for(fut, timeout=30)"), "DLK017": ("fork-with-threads", "medium", "os.fork / 'fork' start method in a module that also starts threads", "fork copies only the calling thread; locks held by other threads (logging, allocator, CUDA,\n" "gRPC, BLAS) stay locked forever in the child. 3.12+ warns with DeprecationWarning.", "Use the 'spawn' or 'forkserver' start method (forkserver is the Linux default since 3.14):\n" " ctx = multiprocessing.get_context(\"spawn\")\n with ctx.Pool(4) as pool:\n ..."), "DLK018": ("lock-in-signal-handler", "medium", "Signal handler acquires a lock or logs", "Python signal handlers run in the main thread between bytecodes, possibly while that thread\n" "already holds the same lock: a non-reentrant lock deadlocks, buffered I/O raises.", "Only set a flag in the handler and do the work elsewhere:\n" " stop = threading.Event()\n signal.signal(signal.SIGTERM, lambda *_: stop.set())\n" "In asyncio: loop.add_signal_handler(signal.SIGTERM, stop_event.set)"), } SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2} PRIMS = {"Lock", "RLock", "Semaphore", "BoundedSemaphore", "Condition", "Event", "Barrier", "Queue", "LifoQueue", "PriorityQueue", "SimpleQueue", "JoinableQueue", "CapacityLimiter"} LOCK_PRIMS = {"Lock", "RLock", "Condition"} SEM_PRIMS = {"Semaphore", "BoundedSemaphore", "CapacityLimiter"} QUEUE_PRIMS = {"Queue", "LifoQueue", "PriorityQueue", "SimpleQueue", "JoinableQueue"} JOINABLE_QUEUES = {"Queue", "LifoQueue", "PriorityQueue", "JoinableQueue"} EXECUTORS = {"ThreadPoolExecutor", "ProcessPoolExecutor", "InterpreterPoolExecutor"} ASYNC_FAMILIES = {"asyncio", "anyio", "trio"} SPAWN_CALLS = {"create_task", "ensure_future", "start_soon", "run_coroutine_threadsafe", "submit", "run_in_executor", "to_thread", "call_soon", "call_soon_threadsafe", "call_later"} LOG_METHODS = {"debug", "info", "warning", "warn", "error", "exception", "critical", "log"} LOCK_NAME = re.compile(r"(?:^|_)(?:r?lock|mutex)(?:_|$|\d)|^r?lock") COND_NAME = re.compile(r"(?:^|_)(?:cond|cv|condition|not_empty|not_full)(?:_|$|\d)") SEM_NAME = re.compile(r"(?:^|_)(?:sem|semaphore|limiter|slots?|permits?)(?:_|$|\d)") QUEUE_NAME = re.compile(r"(?:^|_)q$|queue") FUNC_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef) SCOPE_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda) def dotted(node): """Return 'a.b.c' for Name/Attribute chains, else ''.""" parts = [] while isinstance(node, ast.Attribute): parts.append(node.attr) node = node.value if isinstance(node, ast.Name): parts.append(node.id) return ".".join(reversed(parts)) return "" def last(name): return name.rsplit(".", 1)[-1] def walk_local(nodes): """Walk nodes without descending into nested functions, classes or lambdas.""" stack = list(nodes) while stack: node = stack.pop() yield node for child in ast.iter_child_nodes(node): if not isinstance(child, SCOPE_TYPES): stack.append(child) def walk_no_spawn(node): """Walk an expression, skipping arguments of calls that run code elsewhere (create_task, submit...).""" stack = [node] while stack: n = stack.pop() yield n if isinstance(n, SCOPE_TYPES): continue if isinstance(n, ast.Call) and last(dotted(n.func)) in SPAWN_CALLS: stack.append(n.func) continue stack.extend(ast.iter_child_nodes(n)) def header_exprs(st): """Expressions evaluated by statement `st` itself (not by its nested blocks).""" if isinstance(st, (ast.If, ast.While)): return [st.test] if isinstance(st, (ast.For, ast.AsyncFor)): return [st.iter] if isinstance(st, (ast.With, ast.AsyncWith)): return [i.context_expr for i in st.items] if isinstance(st, SCOPE_TYPES) or type(st).__name__ in ("Try", "TryStar"): return [] if hasattr(st, "cases"): # match statement (3.10+) return [st.subject] return [st] def sub_blocks(st): out = [] for field in ("body", "orelse", "finalbody"): block = getattr(st, field, None) if isinstance(block, list): out.append(block) for h in getattr(st, "handlers", []) or []: out.append(h.body) for c in getattr(st, "cases", []) or []: out.append(c.body) return out def kw(call, name): for k in call.keywords: if k.arg == name: return k.value return None def is_true(node): return isinstance(node, ast.Constant) and node.value is True def is_false(node): return isinstance(node, ast.Constant) and node.value is False class FuncInfo: def __init__(self, node, cls): self.node = node self.cls = cls self.name = node.name self.qual = (cls or "", node.name) self.display = f"{cls}.{node.name}" if cls else node.name self.is_async = isinstance(node, ast.AsyncFunctionDef) self.acq = [] # (lock key, line, held tuple of (key, line)) self.calls = [] # (callee qual, line, held) class Module: """All module-wide facts the rules need.""" def __init__(self, path, source, tree): self.path = path self.lines = source.splitlines() self.tree = tree self.findings = [] self.aliases = {} # local name -> dotted origin self.kinds = {} # normalized key -> (family, prim) self.bases = {} # class name -> base class names (last segment) self.funcs = {} # qual -> FuncInfo self.parents = {} self.has_task_done = False self.resolvers = set() # last segment of receivers of set_result/set_exception/cancel self.threads_started = False self.thread_joins = False for parent in ast.walk(tree): for child in ast.iter_child_nodes(parent): self.parents[child] = parent # ------------------------------------------------------------ helpers def add(self, rule, node, message, fix=None): line = getattr(node, "lineno", 1) src = self.lines[line - 1] if 0 < line <= len(self.lines) else "" m = re.search(r"#\s*noqa(?::\s*([A-Z0-9, ]+))?", src) if m and (not m.group(1) or rule in [c.strip() for c in m.group(1).split(",")]): return slug, sev, _title, why, default_fix = RULES[rule] key = (rule, line, getattr(node, "col_offset", 0)) if any((f["rule"], f["line"], f["col"] - 1) == key for f in self.findings): return self.findings.append({ "path": self.path, "line": line, "col": getattr(node, "col_offset", 0) + 1, "rule": rule, "name": slug, "severity": sev, "message": message, "why": why, "code": src.strip(), "fix": fix or default_fix, }) def resolve(self, name): if not name: return "" head, _, rest = name.partition(".") origin = self.aliases.get(head) if origin: return origin + ("." + rest if rest else "") return name def norm(self, expr, cls): d = dotted(expr) if not d: return "" head, _, rest = d.partition(".") if cls and head in ("self", "cls") and rest: return f"{cls}.{rest}" return d def classify(self, func_expr): """(family, prim) for a constructor reference like threading.Lock / asyncio.Queue.""" name = self.resolve(dotted(func_expr)) if not name: return None parts = name.split(".") prim = parts[-1] fam = parts[0] if prim in EXECUTORS: return ("concurrent", prim) if prim in ("Thread", "Timer") and fam == "threading": return ("threading", prim) if prim == "Process" and fam == "multiprocessing": return ("multiprocessing", prim) if prim == "Future" and fam in ("asyncio", "concurrent"): return (fam, "Future") if prim not in PRIMS or len(parts) < 2: return None if fam in ASYNC_FAMILIES: return ("asyncio", prim) if fam in ("threading", "queue"): return ("threading", prim) if fam == "multiprocessing": return ("multiprocessing", prim) return None def lookup(self, key): """kinds[key], following base classes defined in this module for 'Class.attr' keys.""" if key in self.kinds: return self.kinds[key] head, _, attr = key.partition(".") seen = set() todo = list(self.bases.get(head, ())) while todo and attr: b = todo.pop() if b in seen: continue seen.add(b) if f"{b}.{attr}" in self.kinds: return self.kinds[f"{b}.{attr}"] todo.extend(self.bases.get(b, ())) return None def kind(self, expr, cls): return self.lookup(self.norm(expr, cls)) def is_lockish(self, expr, cls): """Key when expr is a lock usable for ordering analysis, else ''.""" if isinstance(expr, ast.Call): return "" key = self.norm(expr, cls) if not key: return "" k = self.lookup(key) if k: return key if k[1] in LOCK_PRIMS else "" return key if LOCK_NAME.search(last(key).lower()) else "" def reentrant(self, key): k = self.lookup(key) if k: return k[1] == "RLock" or (k[1] == "Condition" and k[0] in ("threading", "multiprocessing")) return "rlock" in last(key).lower() def sync_queue(self, expr, cls): k = self.kind(expr, cls) return bool(k and k[0] in ("threading", "multiprocessing") and k[1] in QUEUE_PRIMS) # ------------------------------------------------------------ pass 1: facts def collect(self): for node in ast.walk(self.tree): if isinstance(node, ast.Import): for a in node.names: if a.asname: self.aliases[a.asname] = a.name elif isinstance(node, ast.ImportFrom) and node.module and not node.level: for a in node.names: self.aliases[a.asname or a.name] = f"{node.module}.{a.name}" for node in ast.walk(self.tree): if isinstance(node, ast.ClassDef): self.bases[node.name] = [last(dotted(b)) for b in node.bases if dotted(b)] self._collect_scope(self.tree.body, None) for node in ast.walk(self.tree): if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): if node.func.attr == "task_done": self.has_task_done = True if node.func.attr in ("set_result", "set_exception", "cancel"): self.resolvers.add(last(dotted(node.func.value))) def _record_assign(self, target, value, cls, in_class_body): if not isinstance(value, ast.Call): return kind = self.classify(value.func) if kind is None and self.resolve(dotted(value.func)) == "multiprocessing.get_context" \ and isinstance(target, ast.Name): self.aliases[target.id] = "multiprocessing" return if kind is None and last(dotted(value.func)) == "create_future": kind = ("asyncio", "Future") if kind is None and last(dotted(value.func)) == "field": fac = kw(value, "default_factory") if fac is not None: kind = self.classify(fac) if kind is None and isinstance(value.func, ast.Attribute) and value.func.attr in PRIMS \ and self.aliases.get(dotted(value.func.value)) == "multiprocessing": kind = ("multiprocessing", value.func.attr) if kind is None: return if in_class_body and isinstance(target, ast.Name): key = f"{cls}.{target.id}" else: key = self.norm(target, cls) if key: self.kinds[key] = kind def _collect_scope(self, stmts, cls, in_class_body=False): for st in stmts: if isinstance(st, ast.ClassDef): self._collect_scope(st.body, st.name, True) continue if isinstance(st, FUNC_TYPES): info = FuncInfo(st, cls if in_class_body else None) self.funcs.setdefault(info.qual, info) a = st.args for arg in getattr(a, "posonlyargs", []) + a.args + a.kwonlyargs: ann = arg.annotation if isinstance(ann, ast.Subscript): # asyncio.Queue[str] ann = ann.value if isinstance(ann, ast.Constant) and isinstance(ann.value, str): try: ann = ast.parse(ann.value, mode="eval").body except SyntaxError: ann = None kind = self.classify(ann) if isinstance(ann, (ast.Name, ast.Attribute)) else None if kind and arg.arg not in self.kinds: self.kinds[arg.arg] = kind self._collect_scope(st.body, cls, False) continue for node in walk_local([st]): if isinstance(node, ast.Assign): for t in node.targets: self._record_assign(t, node.value, cls, in_class_body and node is st) elif isinstance(node, ast.AnnAssign) and node.value is not None: self._record_assign(node.target, node.value, cls, in_class_body and node is st) elif isinstance(node, (ast.With, ast.AsyncWith)): for item in node.items: if item.optional_vars is not None: self._record_assign(item.optional_vars, item.context_expr, cls, False) for block in sub_blocks(st): nested = [s for s in block if isinstance(s, (ast.ClassDef,) + FUNC_TYPES)] if nested: self._collect_scope(nested, cls, False) # ------------------------------------------------------------ pass 2: lock graph def callee(self, call, cls): f = call.func if isinstance(f, ast.Name) and ("", f.id) in self.funcs: return ("", f.id) if isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name) and f.value.id in ("self", "cls") \ and cls and (cls, f.attr) in self.funcs: return (cls, f.attr) return None def scan_block(self, stmts, held, info): held = list(held) cls = info.cls for st in stmts: if isinstance(st, SCOPE_TYPES): continue for expr in header_exprs(st): for n in walk_no_spawn(expr): if not isinstance(n, ast.Call): continue if isinstance(n.func, ast.Attribute) and n.func.attr == "acquire": k = self.is_lockish(n.func.value, cls) if k: info.acq.append((k, n.lineno, tuple(held))) held.append((k, n.lineno)) continue target = self.callee(n, cls) if target: info.calls.append((target, n.lineno, tuple(held))) if isinstance(st, (ast.With, ast.AsyncWith)): inner = list(held) for item in st.items: k = self.is_lockish(item.context_expr, cls) if k: info.acq.append((k, item.context_expr.lineno, tuple(inner))) inner.append((k, item.context_expr.lineno)) self.scan_block(st.body, inner, info) else: for block in sub_blocks(st): self.scan_block(block, held, info) released = {self.is_lockish(n.func.value, cls) for n in walk_local([st]) if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "release"} if released: held = [h for h in held if h[0] not in released] def lock_graph(self): for info in self.funcs.values(): self.scan_block(info.node.body, [], info) owned = {} for q, info in self.funcs.items(): first = {} for k, line, _ in info.acq: first.setdefault(k, line) owned[q] = first edges = {} def edge(a, b, text, node_line): edges.setdefault((a, b), (text, node_line)) for q, info in self.funcs.items(): for k, line, held in info.acq: for hk, hl in held: if hk == k: if not self.reentrant(k): self.add("DLK002", _Pos(line), f"`{k}` is acquired again on line {line} in " f"{info.display}() while already held since line {hl}; a non-reentrant " "lock waits for itself forever") else: edge(hk, k, f"{info.display}() takes {hk} (line {hl}) then {k} (line {line})", line) for target, line, held in info.calls: if not held: continue cinfo = self.funcs[target] for k, kl in owned.get(target, {}).items(): for hk, hl in held: if hk == k: if not self.reentrant(k): self.add("DLK002", _Pos(line), f"{info.display}() holds `{k}` (line {hl}) and " f"calls {cinfo.display}(), which acquires `{k}` again (line {kl}); " "the lock is not reentrant, so this deadlocks") else: edge(hk, k, f"{info.display}() holds {hk} (line {hl}) and calls " f"{cinfo.display}() on line {line}, which takes {k} (line {kl})", line) self.report_cycles(edges) def report_cycles(self, edges): adj = {} for a, b in edges: adj.setdefault(a, set()).add(b) seen = set() nodes = sorted(adj) for start in nodes: stack = [(start, [start])] while stack: node, path = stack.pop() for nxt in sorted(adj.get(node, ())): if nxt == start and len(path) >= 2: cyc = path[:] canon = tuple(sorted(cyc)) if canon in seen: continue seen.add(canon) pairs = list(zip(cyc, cyc[1:] + [cyc[0]])) evid = [edges[p] for p in pairs] line = max(e[1] for e in evid) chain = " -> ".join(cyc + [cyc[0]]) self.add("DLK001", _Pos(line), f"lock-order cycle {chain}: " + "; ".join(e[0] for e in evid)) elif nxt not in path and nxt > start and len(path) < 5: stack.append((nxt, path + [nxt])) # ------------------------------------------------------------ pass 3: per-node rules def check(self): self.collect() self.lock_graph() Checker(self).visit(self.tree) self.check_threads() self.check_executor_self_wait() self.check_fork() self.check_signal_handlers() self.check_futures() for info in self.funcs.values(): self.check_acquire_finally(info.node.body, info, set()) self.check_process_drain(info) self.check_bounded_put(info) return self.findings def check_acquire_finally(self, stmts, info, guarded): cls = info.cls if info.name in ("__enter__", "__aenter__", "acquire", "__exit__", "__aexit__", "release"): return released_here = {self.norm(n.func.value, cls) for n in walk_local(info.node.body) if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "release"} for i, st in enumerate(stmts): if isinstance(st, SCOPE_TYPES): continue for expr in header_exprs(st): for n in walk_local([expr]): if not (isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "acquire"): continue recv = n.func.value key = self.norm(recv, cls) k = self.lookup(key) lockish = (k and (k[1] in LOCK_PRIMS or k[1] in SEM_PRIMS)) or ( not k and key and (LOCK_NAME.search(last(key).lower()) or SEM_NAME.search(last(key).lower()))) if not lockish or key in guarded or key not in released_here: continue # released elsewhere: ownership handed off (e.g. paired acquire/release methods) later = stmts[i + 1:] if isinstance(st, ast.If): later = st.body + st.orelse + later if any(self._finally_releases(s, key, cls) for s in later): continue if later and self._is_release(later[0], key, cls): continue # released immediately: nothing in between can raise self.add("DLK009", n, f"`{dotted(recv)}.acquire()` in {info.display}() has no " f"`{dotted(recv)}.release()` in a finally block: any exception leaks the " "lock/permit and later callers wait forever") if type(st).__name__ in ("Try", "TryStar"): rel = {self.norm(n.func.value, cls) for n in walk_local(st.finalbody) if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "release"} self.check_acquire_finally(st.body, info, guarded | rel) for h in st.handlers: self.check_acquire_finally(h.body, info, guarded) self.check_acquire_finally(st.orelse, info, guarded | rel) self.check_acquire_finally(st.finalbody, info, guarded) else: for block in sub_blocks(st): self.check_acquire_finally(block, info, guarded) def _is_release(self, st, key, cls): return isinstance(st, ast.Expr) and isinstance(st.value, ast.Call) \ and isinstance(st.value.func, ast.Attribute) and st.value.func.attr == "release" \ and self.norm(st.value.func.value, cls) == key def _finally_releases(self, st, key, cls): if type(st).__name__ not in ("Try", "TryStar"): return False blocks = list(st.finalbody) for h in st.handlers: # `except BaseException: lock.release(); raise` also covers every error path if h.type is None or last(dotted(h.type)) == "BaseException": blocks.extend(h.body) return any(isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "release" and self.norm(n.func.value, cls) == key for n in walk_local(blocks)) def check_threads(self): thread_join_any = False for n in ast.walk(self.tree): if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "join": if not n.args or kw(n, "timeout") is not None or ( len(n.args) == 1 and isinstance(n.args[0], ast.Constant) and isinstance(n.args[0].value, (int, float))): if not isinstance(n.func.value, ast.Constant): thread_join_any = True for info_cls, call in self._calls_with_class(): kind = self.classify(call.func) if not kind or kind[1] != "Thread": continue self.threads_started = True if is_true(kw(call, "daemon")): continue parent = self.parents.get(call) if isinstance(parent, ast.Attribute) and parent.attr == "start": self.add("DLK013", call, "non-daemon Thread is started without keeping a handle, so it can " "never be joined; interpreter exit waits for it forever if it blocks") continue if isinstance(parent, (ast.Assign, ast.AnnAssign)): target = parent.targets[0] if isinstance(parent, ast.Assign) else parent.target key = self.norm(target, info_cls) if not key: continue if self._daemon_set(key, info_cls) or self._joined(key, info_cls): continue if not self._started(key, info_cls): continue esc = self._escapes(key, info_cls) if esc == "handed-off" or (esc and thread_join_any): continue self.add("DLK013", call, f"non-daemon Thread `{key}` is started but never joined; a worker " "blocked on a queue or lock makes interpreter shutdown hang") elif not thread_join_any: self.add("DLK013", call, "non-daemon Threads are created here but nothing in the module " "joins a thread; shutdown waits for them forever if one blocks") def _calls_with_class(self): def rec(node, cls): for child in ast.iter_child_nodes(node): c = child.name if isinstance(child, ast.ClassDef) else cls if isinstance(child, ast.Call): yield cls, child yield from rec(child, c) yield from rec(self.tree, None) def _method_calls(self, key, attr, cls_hint): for c, n in self._calls_with_class(): if isinstance(n.func, ast.Attribute) and n.func.attr == attr and self.norm(n.func.value, c) == key: yield n def _daemon_set(self, key, cls): for n in ast.walk(self.tree): if isinstance(n, ast.Assign) and is_true(n.value): for t in n.targets: if isinstance(t, ast.Attribute) and t.attr == "daemon" and \ (self.norm(t.value, cls) == key or dotted(t.value) == key): return True return any(n.args and is_true(n.args[0]) for n in self._method_calls(key, "setDaemon", cls)) def _joined(self, key, cls): return any(True for _ in self._method_calls(key, "join", cls)) def _started(self, key, cls): return any(True for _ in self._method_calls(key, "start", cls)) def _escapes(self, key, cls): name = last(key) for n in ast.walk(self.tree): if isinstance(n, (ast.Name, ast.Attribute)) and isinstance(getattr(n, "ctx", None), ast.Load) \ and (dotted(n) == key or (isinstance(n, ast.Name) and n.id == name)): parent = self.parents.get(n) if isinstance(parent, ast.Attribute): continue if isinstance(parent, (ast.Return, ast.Yield, ast.YieldFrom)): return "handed-off" # the caller owns the thread now return "escaped" return "" def check_executor_self_wait(self): submitted = {} for c, n in self._calls_with_class(): if not isinstance(n.func, ast.Attribute): continue if n.func.attr in ("submit", "map") and n.args: pool, fn = n.func.value, n.args[0] elif n.func.attr == "run_in_executor" and len(n.args) >= 2: pool, fn = n.args[0], n.args[1] else: continue pkey = self.norm(pool, c) k = self.kinds.get(pkey) if not k or k[1] != "ThreadPoolExecutor": continue target = None if isinstance(fn, ast.Name) and ("", fn.id) in self.funcs: target = ("", fn.id) elif isinstance(fn, ast.Attribute) and isinstance(fn.value, ast.Name) and fn.value.id == "self" \ and c and (c, fn.attr) in self.funcs: target = (c, fn.attr) if target: submitted.setdefault(target, set()).add(pkey) for target, pools in submitted.items(): info = self.funcs[target] nodes = list(walk_local(info.node.body)) waits = any(isinstance(n, ast.Call) and ( (isinstance(n.func, ast.Attribute) and n.func.attr == "result") or last(self.resolve(dotted(n.func))) in ("wait", "as_completed") or (isinstance(n.func, ast.Name) and n.func.id in ("list", "tuple", "sorted") and n.args and isinstance(n.args[0], ast.Call) and isinstance(n.args[0].func, ast.Attribute) and n.args[0].func.attr == "map")) for n in nodes) or any(isinstance(n, (ast.For, ast.comprehension)) and isinstance(n.iter, ast.Call) and isinstance(n.iter.func, ast.Attribute) and n.iter.func.attr == "map" for n in nodes) if not waits: continue for n in nodes: if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) \ and n.func.attr in ("submit", "map") and self.norm(n.func.value, info.cls) in pools: self.add("DLK012", n, f"{info.display}() runs inside executor `{dotted(n.func.value)}` and " "submits more work to the same executor, then waits for it: when every worker " "is busy in an outer call, the inner tasks can never start") break def check_fork(self): forks = [] for n in ast.walk(self.tree): if not isinstance(n, ast.Call): continue name = self.resolve(dotted(n.func)) if name in ("os.fork", "os.forkpty"): forks.append(n) elif last(name) in ("set_start_method", "get_context") and n.args \ and isinstance(n.args[0], ast.Constant) and n.args[0].value == "fork": forks.append(n) if not forks: return threaded = False for n in ast.walk(self.tree): if isinstance(n, ast.Call): k = self.classify(n.func) if k and k[1] in ("Thread", "Timer", "ThreadPoolExecutor"): threaded = True if last(self.resolve(dotted(n.func))) in ("to_thread", "run_in_executor"): threaded = True if threaded: for n in forks: self.add("DLK017", n, "fork in a process that also runs threads: a lock held by another " "thread at fork time (logging handler, allocator, CUDA, gRPC, a queue) stays locked " "forever in the child") def check_signal_handlers(self): for c, n in self._calls_with_class(): if self.resolve(dotted(n.func)) != "signal.signal" or len(n.args) < 2: continue h = n.args[1] target = None if isinstance(h, ast.Name) and ("", h.id) in self.funcs: target = ("", h.id) elif isinstance(h, ast.Attribute) and isinstance(h.value, ast.Name) and h.value.id == "self" \ and c and (c, h.attr) in self.funcs: target = (c, h.attr) if not target: continue info = self.funcs[target] for m in walk_local(info.node.body): bad = None if isinstance(m, ast.With): for item in m.items: if self.is_lockish(item.context_expr, info.cls): bad = f"takes `{dotted(item.context_expr)}`" elif isinstance(m, ast.Call) and isinstance(m.func, ast.Attribute): recv = dotted(m.func.value) if m.func.attr == "acquire" and self.is_lockish(m.func.value, info.cls): bad = f"acquires `{recv}`" elif m.func.attr in LOG_METHODS and ("log" in recv.lower()): bad = f"calls `{recv}.{m.func.attr}()` (logging handlers hold locks)" if bad: self.add("DLK018", m, f"signal handler {info.display}() {bad}; it runs in the main thread " "between bytecodes, possibly while that thread already holds the same lock") break def check_futures(self): futs = {k for k, v in self.kinds.items() if v[1] == "Future"} if not futs: return for c, n in self._awaits_with_class(): key = self.norm(n.value, c) if key not in futs or last(key) in self.resolvers: continue if self._future_escapes(key): continue self.add("DLK016", n, f"`await {dotted(n.value)}` waits on a Future that nothing in this module " "resolves (no set_result/set_exception/cancel, never handed to another object)") def _awaits_with_class(self): def rec(node, cls): for child in ast.iter_child_nodes(node): c = child.name if isinstance(child, ast.ClassDef) else cls if isinstance(child, ast.Await): yield cls, child yield from rec(child, c) yield from rec(self.tree, None) def _future_escapes(self, key): name = last(key) for c, node in self._loads_with_class(): if self.norm(node, c) != key and not (isinstance(node, ast.Name) and node.id == name): continue parent = self.parents.get(node) if isinstance(parent, (ast.Attribute, ast.Await)): continue return True return False def _loads_with_class(self): def rec(node, cls): for child in ast.iter_child_nodes(node): c = child.name if isinstance(child, ast.ClassDef) else cls if isinstance(child, (ast.Name, ast.Attribute)) and isinstance(child.ctx, ast.Load): yield cls, child yield from rec(child, c) yield from rec(self.tree, None) def check_process_drain(self, info): cls = info.cls nodes = sorted((n for n in walk_local(info.node.body) if isinstance(n, ast.Call)), key=lambda n: (n.lineno, n.col_offset)) mp_queues = {k for k, v in self.kinds.items() if v[0] == "multiprocessing" and v[1] in QUEUE_PRIMS} if not mp_queues: return has_proc = any((self.classify(n.func) or ("", ""))[1] == "Process" or (isinstance(n.func, ast.Attribute) and n.func.attr == "Process") for n in nodes) if not has_proc: return first_join = first_get = None for n in nodes: if not isinstance(n.func, ast.Attribute): continue key = self.norm(n.func.value, cls) if n.func.attr == "join" and key not in mp_queues and not isinstance(n.func.value, ast.Constant) \ and len(n.args) <= 1 and first_join is None: first_join = n if n.func.attr in ("get", "get_nowait") and key in mp_queues and first_get is None: first_get = n if first_join is not None and first_get is not None and \ (first_join.lineno, first_join.col_offset) < (first_get.lineno, first_get.col_offset): self.add("DLK014", first_join, f"`{dotted(first_join.func.value)}.join()` runs before " f"`{dotted(first_get.func.value)}.get()` (line {first_get.lineno}) drains the queue: a child " "with unflushed queue data cannot exit, so join() waits forever") def check_bounded_put(self, info): cls = info.cls bounded = {} for n in walk_local(info.node.body): if isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name) \ and isinstance(n.value, ast.Call): k = self.classify(n.value.func) if not k or k[1] not in JOINABLE_QUEUES: continue size = kw(n.value, "maxsize") or (n.value.args[0] if n.value.args else None) if isinstance(size, ast.Constant) and isinstance(size.value, int) and size.value > 0: bounded[n.targets[0].id] = n for name in bounded: puts, reads, escaped = [], False, False for n in ast.walk(info.node): if isinstance(n, ast.Name) and n.id == name and isinstance(n.ctx, ast.Load): parent = self.parents.get(n) if isinstance(parent, ast.Attribute): if parent.attr in ("put", "put_nowait"): puts.append(parent) elif parent.attr.startswith("get"): reads = True else: escaped = True if puts and not reads and not escaped: self.add("DLK015", puts[0], f"bounded queue `{name}` is filled here but nothing in " f"{info.display}() reads it or hands it to a consumer: put() blocks forever once full") class _Pos: def __init__(self, line): self.lineno = line self.col_offset = 0 class Checker(ast.NodeVisitor): """Rules that depend on the enclosing function (async or not) and loop nesting.""" def __init__(self, mod): self.m = mod self.scopes = [(False, None, 0)] # (is_async, cls, while depth) self.cls = None self.awaited = set() self.threadsafe_futs = [] self.submit_futs = [] @property def in_async(self): return self.scopes[-1][0] def visit_ClassDef(self, node): saved, self.cls = self.cls, node.name self.generic_visit(node) self.cls = saved def _func(self, node, is_async): self.scopes.append((is_async, self.cls, 0)) saved = (self.threadsafe_futs, self.submit_futs) self.threadsafe_futs, self.submit_futs = set(), set() for n in walk_local(node.body): if isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.value, ast.Call): fname = last(self.m.resolve(dotted(n.value.func))) key = self.m.norm(n.targets[0], self.cls) if fname == "run_coroutine_threadsafe": self.threadsafe_futs.add(key) elif fname == "submit": self.submit_futs.add(key) if isinstance(n, (ast.For, ast.comprehension)) and isinstance(n.iter, ast.Call) \ and self.m.resolve(dotted(n.iter.func)) in ("concurrent.futures.as_completed",): self.submit_futs.add(dotted(n.target)) self.generic_visit(node) self.threadsafe_futs, self.submit_futs = saved self.scopes.pop() def visit_FunctionDef(self, node): self._func(node, False) def visit_AsyncFunctionDef(self, node): self._func(node, True) def visit_Lambda(self, node): self.scopes.append((False, self.cls, 0)) self.generic_visit(node) self.scopes.pop() def visit_While(self, node): a, c, d = self.scopes[-1] self.visit(node.test) self.scopes[-1] = (a, c, d + 1) for s in node.body: self.visit(s) self.scopes[-1] = (a, c, d) for s in node.orelse: self.visit(s) def visit_Await(self, node): if isinstance(node.value, ast.Call): self.awaited.add(id(node.value)) self.generic_visit(node) def visit_With(self, node): if self.in_async: for item in node.items: e = item.context_expr if isinstance(e, ast.Call): continue k = self.m.kind(e, self.cls) if (k and k[0] in ("threading", "multiprocessing") and (k[1] in LOCK_PRIMS or k[1] in SEM_PRIMS)) \ or (not k and LOCK_NAME.search(last(dotted(e)).lower() or "-")): held_await = any(isinstance(n, ast.Await) for n in walk_local(node.body)) extra = " and awaits while holding it (other tasks on this loop can never release it)" \ if held_await else "" self.m.add("DLK003", e, f"`with {dotted(e)}:` takes a threading lock inside async def" f"{extra}; a contended acquire blocks the whole event loop") self.generic_visit(node) def visit_Call(self, node): m, cls = self.m, self.cls f = node.func name = m.resolve(dotted(f)) attr = f.attr if isinstance(f, ast.Attribute) else "" recv = f.value if isinstance(f, ast.Attribute) else None awaited = id(node) in self.awaited rk = m.kind(recv, cls) if recv is not None else None rkey = m.norm(recv, cls) if recv is not None else "" if self.in_async and not awaited: if name == "time.sleep": m.add("DLK004", node, "time.sleep() inside async def blocks the event loop; every other task " "(streams, heartbeats, timeouts) stops") elif name in ("asyncio.run",) or attr in ("run_until_complete", "run_forever"): m.add("DLK007", node, f"`{dotted(f)}()` inside async def: a loop is already running in this " "thread (RuntimeError, or a deadlock under nest_asyncio)") elif attr == "result": src = recv.func if isinstance(recv, ast.Call) else None src_name = last(m.resolve(dotted(src))) if src is not None else "" if src_name == "run_coroutine_threadsafe" or rkey in self.threadsafe_futs: m.add("DLK006", node, "run_coroutine_threadsafe(...).result() inside async def: when the " "target loop is this loop, it waits for work that can only run after it returns") elif src_name == "submit" or (rkey and rkey in self.submit_futs): m.add("DLK005", node, f"`{ast.unparse(node)}` blocks the event loop until the executor " "finishes; await asyncio.wrap_future(...) instead") elif attr == "join" and (not node.args or kw(node, "timeout") is not None) \ and not (rk and rk[0] == "asyncio") and not isinstance(recv, ast.Constant): m.add("DLK005", node, f"`{dotted(f)}()` inside async def blocks the event loop until the " "thread/process/queue finishes; if it needs this loop, it never will") elif attr in ("get", "put") and m.sync_queue(recv, cls) and not is_false(kw(node, "block")) \ and not (node.args[1:2] and is_false(node.args[1])) \ and not (attr == "get" and node.args and is_false(node.args[0])): m.add("DLK005", node, f"`{dotted(f)}()` on a thread/process queue inside async def blocks " "the event loop; use asyncio.Queue or await asyncio.to_thread(...)") elif attr == "wait" and rk and rk[0] in ("threading", "multiprocessing") \ and rk[1] in ("Event", "Condition", "Barrier"): m.add("DLK005", node, f"`{dotted(f)}()` on a threading {rk[1]} inside async def blocks the " "event loop; use asyncio.Event or await asyncio.to_thread(...)") elif name in ("concurrent.futures.wait",): m.add("DLK005", node, "concurrent.futures.wait() inside async def blocks the event loop; use " "await asyncio.wait([asyncio.wrap_future(f) for f in futs])") elif attr == "acquire" and rk and rk[0] in ("threading", "multiprocessing"): m.add("DLK003", node, f"`{dotted(f)}()` acquires a threading lock inside async def; a " "contended acquire blocks the whole event loop") if attr == "join" and m.has_task_done is False and ( (rk and rk[1] in JOINABLE_QUEUES) or (not rk and recv is not None and not node.args and QUEUE_NAME.search(last(dotted(recv)).lower() or "-"))): m.add("DLK008", node, f"`{dotted(f)}()` waits for every item to be marked done, but nothing in this " "module calls task_done(): it never returns") if attr == "wait" and recv is not None and self.scopes[-1][2] == 0 and not node.args \ and kw(node, "timeout") is None: is_cond = (rk and rk[1] == "Condition") or (not rk and COND_NAME.search(last(dotted(recv)).lower() or "-")) if is_cond: m.add("DLK010", node, f"`{dotted(f)}()` is not inside a `while not predicate:` loop: a " "spurious, stolen or early notify leaves the waiter acting on stale state or " "waiting forever; use wait_for(predicate)") if not self.in_async and not awaited and rk and rk[0] in ("threading", "multiprocessing"): no_timeout = kw(node, "timeout") is None if attr == "get" and rk[1] in QUEUE_PRIMS and no_timeout and not node.args[:1] \ and not is_false(kw(node, "block")): m.add("DLK011", node, f"`{dotted(f)}()` blocks forever if the producer dies; pass " "timeout= and re-check a stop flag (or use Queue.shutdown() on 3.13+)") elif attr == "wait" and rk[1] in ("Event", "Condition", "Barrier") and no_timeout and not node.args \ and not (rk[1] == "Condition" and self.scopes[-1][2] == 0): m.add("DLK011", node, f"`{dotted(f)}()` without a timeout blocks forever if nobody ever " "sets/notifies it") elif attr == "join" and rk[1] in JOINABLE_QUEUES and m.has_task_done: m.add("DLK011", node, f"`{dotted(f)}()` has no timeout parameter; if a worker dies before " "task_done() it waits forever. Join the workers with a timeout instead, or poll " "unfinished_tasks with a deadline") self.generic_visit(node) # ---------------------------------------------------------------- driver def scan_source(source, path): tree = ast.parse(source, filename=path) return Module(path, source, tree).check() def expand(paths): skip = {".git", ".venv", "venv", "__pycache__", "node_modules", ".tox", ".mypy_cache", "build", "dist"} for p in paths: if p != "-" and os.path.isdir(p): for root, dirs, files in os.walk(p): dirs[:] = sorted(d for d in dirs if d not in skip) for f in sorted(files): if f.endswith(".py"): yield os.path.join(root, f), True else: yield p, False def parse_rule_list(values): out = set() for v in values or []: for tok in v.split(","): tok = tok.strip() if not tok: continue matched = [rid for rid, meta in RULES.items() if rid.startswith(tok.upper()) or meta[0] == tok] if not matched: raise ValueError(f"unknown rule {tok!r}; run --list-rules") out.update(matched) return out def list_rules(): for rid, (slug, sev, title, why, fix) in RULES.items(): print(f"{rid} {slug} [{sev}]\n {title}\n why: " + why.replace("\n", "\n ")) print(" fix: " + fix.replace("\n", "\n ") + "\n") def main(argv=None): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("paths", nargs="*", help="Python files or directories, or - for stdin") ap.add_argument("--json", action="store_true", help="emit machine-readable JSON") ap.add_argument("--strict", action="store_true", help="exit 1 if any finding is reported") ap.add_argument("--select", action="append", help="only these rules (ids, id prefixes or names; comma list)") ap.add_argument("--ignore", action="append", help="skip these rules (ids, id prefixes or names; comma list)") ap.add_argument("--list-rules", action="store_true", help="print every rule with why/fix and exit") args = ap.parse_args(argv) if args.list_rules: list_rules() return 0 if not args.paths: ap.error("at least one path (or -) is required") try: select = parse_rule_list(args.select) or set(RULES) ignore = parse_rule_list(args.ignore) except ValueError as err: ap.error(str(err)) findings, errors, warnings, files = [], [], [], 0 for path, walked in expand(args.paths): label = "" if path == "-" else path try: if path == "-": source = sys.stdin.read() else: with open(path, encoding="utf-8") as fh: source = fh.read() findings.extend(scan_source(source, label)) files += 1 except SyntaxError as err: msg = f"{label}:{err.lineno}: cannot parse: {err.msg}" (warnings if walked else errors).append(msg) except (OSError, UnicodeDecodeError, ValueError) as err: msg = f"{label}: cannot read: {err}" (warnings if walked else errors).append(msg) findings = [f for f in findings if f["rule"] in select and f["rule"] not in ignore] findings.sort(key=lambda f: (f["path"], f["line"], f["col"], f["rule"])) counts = {s: sum(f["severity"] == s for f in findings) for s in SEVERITY_ORDER} if args.json: print(json.dumps({"files": files, "summary": counts, "errors": errors, "warnings": warnings, "findings": findings}, indent=2)) else: for w in warnings: print(f"warning: {w} (skipped)", file=sys.stderr) for e in errors: print(f"error: {e}", file=sys.stderr) if not findings: if files: print(f"No hang risks found in {files} file(s).") else: print(f"{len(findings)} finding(s) in {files} file(s): {counts['high']} high, " f"{counts['medium']} medium, {counts['low']} low\n") for f in findings: print(f"{f['path']}:{f['line']}:{f['col']}: {f['rule']} [{f['severity']}] {f['name']}: " f"{f['message']}") if f["code"]: print(f" > {f['code']}") print(" fix: " + f["fix"].replace("\n", "\n ")) print() if errors: return 2 return 1 if args.strict and findings else 0 if __name__ == "__main__": sys.exit(main())