Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442#!/usr/bin/env python3# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""HTTP API conventions lint.
Static check for the response/error conventions documented in``docs/CONVEY.md`` § "HTTP API conventions". It scans every convey routemodule — discovered by *which files register a Flask ``Blueprint``* under``solstone/apps/`` and ``solstone/convey/`` (so a new app or core blueprint ispicked up automatically) — and, inside each JSON-returning handler, flags theescape hatches the conventions forbid:
- ``abort`` — an ``abort(...)`` call. - ``bare-return`` — a bare ``return "", <4xx>`` tuple. - ``inline-error`` — an in-band error body built inline (``jsonify({"error": ...})`` / ``return {"error": ...}``) outside the sanctioned ``error_response`` / ``error_response_with_reason`` helpers. - ``bare-array`` — returning a bare top-level list (``jsonify([...])`` / ``jsonify(<list-var>)`` / ``return [...]``). - ``render-template`` — a Flask ``render_template(...)`` call anywhere under the scanned source scopes, excluding tests. The only allowed instances are the PDF helpers in the ``news`` and ``reflections`` apps.
A handler is **JSON-governed** when any of its return paths produces a JSONbody: it calls ``jsonify(...)``, returns one of the response helpers(``error_response`` / ``error_response_with_reason`` / ``success_response`` /``respond_collection`` / ``created``), or returns a dict/list literal. A handlerwhose only returns are ``render_template`` / ``redirect`` / ``send_file`` / aplain string (a page-or-file route) is exempt from the ``abort`` and``bare-return`` rules; a bare top-level array is always flagged. Classificationis by return style, never by URL — the ``/api/`` path segment is an unreliablesignal (app routes carry it in the decorator, ``chat_bp`` carries it in theblueprint ``url_prefix``, and genuinely-JSON endpoints carry it nowhere).
The check ships green via a committed ``ALLOWLIST`` of the violations that existon the current tree, keyed by ``(file, kind)`` with an allowed **count**. A newviolation that pushes any ``(file, kind)`` count above its allowed number failsthe check; fixing occurrences lets the allowed count be lowered, so theallowlist ratchets toward empty. It is never keyed by line number (brittle toedits above the occurrence) and never a blanket per-file disable (which wouldhide future new violations in that file).
Exit codes: 0 — no un-allowlisted violations 1 — a (file, kind) count exceeds its allowlisted number"""
from __future__ import annotations
import argparseimport astimport sysfrom pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
# Trees scanned for Blueprint-registering route modules.SCAN_SCOPES: tuple[str, ...] = ( "solstone/apps", "solstone/convey",)
# Response helpers that legitimately produce a JSON body. A call to one of# these is JSON-producing for classification and is NOT an inline error.RESPONSE_HELPERS: frozenset[str] = frozenset( { "error_response", "error_response_with_reason", "success_response", "respond_collection", "created", })
# Decorator attributes that mark a function as a Flask route handler.ROUTE_DECORATORS: frozenset[str] = frozenset( {"route", "get", "post", "put", "patch", "delete"})
# Committed allowlist of violations on the current tree, keyed by# (posix-relative-path, kind) -> allowed count. Ratchets toward empty: lower a# count as occurrences are fixed; never raise one to admit a new violation.ALLOWLIST: dict[tuple[str, str], int] = { ("solstone/apps/network/routes.py", "abort"): 1, ("solstone/apps/news/routes.py", "render-template"): 1, ("solstone/apps/reflections/routes.py", "render-template"): 1,}
def _func_name(func: ast.expr) -> str | None: """Return the called name for ``Name``/``Attribute`` call targets.""" if isinstance(func, ast.Name): return func.id if isinstance(func, ast.Attribute): return func.attr return None
def _iter_local_nodes(node: ast.AST): """Yield descendants of ``node``, not descending into nested functions.
Stops at nested ``FunctionDef`` / ``AsyncFunctionDef`` / ``Lambda`` so that a handler's classification and violations are not polluted by inner helpers or generators (e.g. an SSE ``generate()`` closure). """ for child in ast.iter_child_nodes(node): if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): continue yield child yield from _iter_local_nodes(child)
def _is_route_handler(func: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: for dec in func.decorator_list: if isinstance(dec, ast.Call) and _func_name(dec.func) in ROUTE_DECORATORS: return True return False
def _is_jsonify_call(expr: ast.expr | None) -> bool: return isinstance(expr, ast.Call) and _func_name(expr.func) == "jsonify"
def _collect_list_names(nodes: list[ast.AST]) -> set[str]: """Names bound to an obvious list value within the handler body.""" names: set[str] = set() for node in nodes: if isinstance(node, ast.Assign): value = node.value is_list = isinstance(value, (ast.List, ast.ListComp)) or ( isinstance(value, ast.Call) and _func_name(value.func) == "list" ) if is_list: for target in node.targets: if isinstance(target, ast.Name): names.add(target.id) elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): ann = node.annotation ann_name = None if isinstance(ann, ast.Name): ann_name = ann.id elif isinstance(ann, ast.Subscript) and isinstance(ann.value, ast.Name): ann_name = ann.value.id if ann_name in ("list", "List"): names.add(node.target.id) if isinstance(node.value, (ast.List, ast.ListComp)): names.add(node.target.id) return names
def _is_list_payload(expr: ast.expr | None, list_names: set[str]) -> bool: if isinstance(expr, (ast.List, ast.ListComp)): return True return isinstance(expr, ast.Name) and expr.id in list_names
def _payload_expr(ret: ast.Return) -> ast.expr | None: """The value carried by a return, unwrapping ``(body, status)`` tuples.""" value = ret.value if isinstance(value, ast.Tuple) and value.elts: return value.elts[0] return value
def _produces_json(expr: ast.expr | None, list_names: set[str]) -> bool: if expr is None: return False if _is_jsonify_call(expr): return True if isinstance(expr, ast.Call) and _func_name(expr.func) in RESPONSE_HELPERS: return True if isinstance(expr, (ast.Dict, ast.List, ast.ListComp, ast.DictComp)): return True return _is_list_payload(expr, list_names)
def _dict_has_error_key(expr: ast.expr | None) -> bool: return isinstance(expr, ast.Dict) and any( isinstance(key, ast.Constant) and key.value == "error" for key in expr.keys )
def _is_inline_error(payload: ast.expr | None) -> bool: if _is_jsonify_call(payload): assert isinstance(payload, ast.Call) return len(payload.args) == 1 and _dict_has_error_key(payload.args[0]) return _dict_has_error_key(payload)
def _is_bare_array(payload: ast.expr | None, list_names: set[str]) -> bool: if _is_jsonify_call(payload): assert isinstance(payload, ast.Call) return len(payload.args) == 1 and _is_list_payload(payload.args[0], list_names) return _is_list_payload(payload, list_names)
def _is_bare_status_return(ret: ast.Return) -> bool: value = ret.value if not (isinstance(value, ast.Tuple) and len(value.elts) == 2): return False body, status = value.elts return ( isinstance(body, ast.Constant) and isinstance(body.value, str) and isinstance(status, ast.Constant) and isinstance(status.value, int) and not isinstance(status.value, bool) and 400 <= status.value <= 499 )
def classify_handler( func: ast.FunctionDef | ast.AsyncFunctionDef,) -> list[tuple[int, str]]: """Return ``(lineno, kind)`` violations for a single route handler.""" nodes = list(_iter_local_nodes(func)) list_names = _collect_list_names(nodes)
returns = [n for n in nodes if isinstance(n, ast.Return)] json_governed = any(_produces_json(_payload_expr(r), list_names) for r in returns)
findings: list[tuple[int, str]] = []
# abort(...) anywhere in the handler's own scope. if json_governed: for node in nodes: if isinstance(node, ast.Call) and _func_name(node.func) == "abort": findings.append((node.lineno, "abort"))
for ret in returns: payload = _payload_expr(ret) if json_governed and _is_bare_status_return(ret): findings.append((ret.lineno, "bare-return")) if _is_inline_error(payload): findings.append((ret.lineno, "inline-error")) if _is_bare_array(payload, list_names): findings.append((ret.lineno, "bare-array"))
return findings
def module_registers_blueprint(tree: ast.AST) -> bool: for node in ast.walk(tree): if isinstance(node, ast.Call) and _func_name(node.func) == "Blueprint": return True return False
def discover_modules(root: Path) -> list[Path]: """Posix-relative paths of Blueprint-registering modules under the scopes.""" found: list[Path] = [] for scope in SCAN_SCOPES: scope_dir = root / scope if not scope_dir.is_dir(): continue for path in sorted(scope_dir.rglob("*.py")): if "__pycache__" in path.parts: continue try: tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) except (SyntaxError, UnicodeDecodeError): continue if module_registers_blueprint(tree): found.append(path.relative_to(root)) return found
def _flask_render_template_names(tree: ast.AST) -> tuple[set[str], set[str]]: """Local names bound to flask.render_template.
Returns (direct, modules): ``direct`` are names from ``from flask import render_template [as X]`` (module- OR function-scoped); ``modules`` are names bound to the flask module via ``import flask [as X]``, whose ``.render_template`` attribute is the same callable. """ direct: set[str] = set() modules: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module == "flask": for alias in node.names: if alias.name == "render_template": direct.add(alias.asname or alias.name) elif isinstance(node, ast.Import): for alias in node.names: if alias.name == "flask": modules.add(alias.asname or "flask") return direct, modules
def scan_render_templates(source: str, filename: str = "<source>") -> list[int]: """Line numbers of flask ``render_template(...)`` calls in a module source.""" tree = ast.parse(source, filename=filename) direct, modules = _flask_render_template_names(tree) if not direct and not modules: return [] linenos: list[int] = [] for node in ast.walk(tree): if not isinstance(node, ast.Call): continue func = node.func if isinstance(func, ast.Name) and func.id in direct: linenos.append(node.lineno) elif ( isinstance(func, ast.Attribute) and func.attr == "render_template" and isinstance(func.value, ast.Name) and func.value.id in modules ): linenos.append(node.lineno) return sorted(linenos)
def discover_all_modules(root: Path) -> list[Path]: """Posix-relative paths of ALL non-test ``*.py`` modules under the scopes.""" found: list[Path] = [] for scope in SCAN_SCOPES: scope_dir = root / scope if not scope_dir.is_dir(): continue for path in sorted(scope_dir.rglob("*.py")): if "__pycache__" in path.parts or "tests" in path.parts: continue found.append(path.relative_to(root)) return found
def scan_source(source: str, filename: str = "<source>") -> list[tuple[int, str, str]]: """Return ``(lineno, kind, function_name)`` violations for a module source.""" tree = ast.parse(source, filename=filename) findings: list[tuple[int, str, str]] = [] for node in ast.walk(tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue if not _is_route_handler(node): continue for lineno, kind in classify_handler(node): findings.append((lineno, kind, node.name)) findings.sort() return findings
def scan_file(path: Path) -> list[tuple[int, str, str]]: return scan_source(path.read_text(encoding="utf-8"), filename=str(path))
def count_violations(root: Path) -> dict[tuple[str, str], int]: """Map ``(posix-relpath, kind)`` -> occurrence count across the tree.""" counts: dict[tuple[str, str], int] = {} for rel in discover_modules(root): for _lineno, kind, _func in scan_file(root / rel): counts[(rel.as_posix(), kind)] = counts.get((rel.as_posix(), kind), 0) + 1 for rel in discover_all_modules(root): linenos = scan_render_templates( (root / rel).read_text(encoding="utf-8"), str(rel) ) if linenos: key = (rel.as_posix(), "render-template") counts[key] = counts.get(key, 0) + len(linenos) return counts
def _account( new: list[str], tracked: list[str], rel_str: str, kind: str, linenos: list[int], allowlist: dict[tuple[str, str], int],) -> None: count = len(linenos) allowed = allowlist.get((rel_str, kind), 0) if count > allowed: lines = ", ".join(str(n) for n in sorted(linenos)) new.append(f"{rel_str}: {count} {kind} (allowed {allowed}) at line(s) {lines}") elif allowed: tracked.append(f"{rel_str}: {count}/{allowed} {kind} (allowlisted)")
def evaluate( root: Path, allowlist: dict[tuple[str, str], int],) -> tuple[list[str], list[str]]: """Return ``(new_violations, tracked)`` human-readable lines.""" new: list[str] = [] tracked: list[str] = [] for rel in discover_modules(root): rel_str = rel.as_posix() findings = scan_file(root / rel) by_kind: dict[str, list[int]] = {} for lineno, kind, _func in findings: by_kind.setdefault(kind, []).append(lineno) for kind, linenos in sorted(by_kind.items()): _account(new, tracked, rel_str, kind, linenos, allowlist) for rel in discover_all_modules(root): rel_str = rel.as_posix() linenos = scan_render_templates( (root / rel).read_text(encoding="utf-8"), str(rel) ) if linenos: _account(new, tracked, rel_str, "render-template", linenos, allowlist) return new, tracked
def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="HTTP API conventions lint") parser.add_argument( "--root", type=Path, default=ROOT, help="Repository root to scan (defaults to the checkout root).", ) args = parser.parse_args(argv)
new, tracked = evaluate(args.root, ALLOWLIST)
if tracked: print("api-conventions: known violations (allowlisted, ratcheting down):") for line in tracked: print(f" {line}") print()
if new: print("api-conventions: NEW violations:", file=sys.stderr) for line in new: print(f" {line}", file=sys.stderr) print(file=sys.stderr) print( "See docs/CONVEY.md § HTTP API conventions. Route a collection " "through respond_collection(), a create through created(), and " "every error through error_response().", file=sys.stderr, ) return 1
print("api-conventions: pass") return 0
if __name__ == "__main__": raise SystemExit(main())