#!/usr/bin/env python3 """drift — find where your code and your config have drifted apart. Zero dependencies, Python 3.8+. Point it at a project and it looks for the quiet ways code and configuration fall out of sync: R1 unexpected_kwarg calls passing keyword args the callee cannot accept R2 config_drift config keys read but never defined, defined but never read R3 magic_number bare numbers doing a job a named constant should do R4 phantom_name names used but never defined (typos that silently default) Every rule comes from a bug I actually shipped or fixed. See README.md. """ from __future__ import annotations import argparse import ast import json import os import re import sys from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Set, Tuple VERSION = "0.1.0" SKIP_DIRS = { ".git", ".hg", ".svn", "__pycache__", "node_modules", "venv", ".venv", ".tox", "dist", "build", "site-packages", ".mypy_cache", ".pytest_cache", } CONFIG_NAMES = {"config", "conf", "settings", "cfg", "env", "defaults", "options"} MAGIC_EXCLUDE = {"0.0", "1.0", "-1.0", "2.0"} MAGIC_MIN_COUNT = 3 ENV_KEY_RE = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*)$") UPPER_NAME_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$") BUILTINS = frozenset("""abs aiter all anext any ascii bin bool breakpoint bytearray bytes callable chr classmethod compile complex delattr dict dir divmod enumerate eval exec filter float format frozenset getattr globals hasattr hash help hex id input int isinstance issubclass iter len list locals map max memoryview min next object oct open ord pow print property range repr reversed round set setattr slice sorted staticmethod str sum super tuple type vars zip BaseException Exception ArithmeticError AssertionError AttributeError BufferError EOFError FloatingPointError GeneratorExit ImportError ModuleNotFoundError IndexError KeyError KeyboardInterrupt LookupError MemoryError NameError NotImplementedError OSError OverflowError RecursionError ReferenceError RuntimeError StopAsyncIteration StopIteration SyntaxError SystemError SystemExit TabError TimeoutError TypeError UnboundLocalError UnicodeDecodeError UnicodeEncodeError UnicodeError UnicodeTranslateError ValueError ZeroDivisionError exit quit copyright credits license Ellipsis NotImplemented""".split()) @dataclass class Finding: rule: str severity: str # error | warning file: str line: int message: str def to_dict(self) -> Dict[str, object]: return { "rule": self.rule, "severity": self.severity, "file": self.file, "line": self.line, "message": self.message, } @dataclass class _Sig: params: Set[str] = field(default_factory=set) has_var_kw: bool = False location: Optional[Tuple[str, int]] = None bases: List[str] = field(default_factory=list) def _iter_project_files(paths: List[str]): for raw in paths: p = Path(raw) if p.is_file(): if p.suffix == ".py" or p.name == ".env": yield p elif p.is_dir(): for f in sorted(p.rglob("*")): if not f.is_file(): continue if any(part in SKIP_DIRS for part in f.parts): continue if f.suffix == ".py" or f.name == ".env": yield f def _parse(p: Path) -> Optional[ast.Module]: try: return ast.parse(p.read_text(encoding="utf-8", errors="replace"), filename=str(p)) except SyntaxError: return None def _sig_from_args(a: ast.arguments) -> _Sig: params = {x.arg for x in a.args} | {x.arg for x in a.kwonlyargs} return _Sig(params=params, has_var_kw=a.kwarg is not None) def _init_sig(cls_node: ast.ClassDef) -> Optional[_Sig]: for n in cls_node.body: if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == "__init__": return _sig_from_args(n.args) return None def _recv_name(node) -> Optional[str]: if isinstance(node, ast.Name): return node.id if isinstance(node, ast.Attribute): return node.attr return None def _add_target_names(node, into: Set[str]) -> None: if isinstance(node, ast.Name): into.add(node.id) elif isinstance(node, (ast.Tuple, ast.List)): for e in node.elts: _add_target_names(e, into) elif isinstance(node, ast.Starred): _add_target_names(node.value, into) def _walk_with_parent(tree): stack = [(tree, None)] while stack: node, parent = stack.pop() yield node, parent for child in reversed(list(ast.iter_child_nodes(node))): stack.append((child, node)) def analyze(paths: List[str], rules: str = "R1,R2,R3,R4") -> Tuple[List[Finding], List[str]]: """Run drift over paths. Returns (findings, notes).""" wanted = set(rules.split(",")) findings: List[Finding] = [] notes: List[str] = [] files = list(_iter_project_files(paths)) trees: Dict[Path, ast.Module] = {} for p in files: if p.suffix != ".py": continue t = _parse(p) if t is None: notes.append(f"drift: skipping {p} (could not parse)") continue trees[p] = t class_sigs: Dict[str, _Sig] = {} func_sigs: Dict[str, _Sig] = {} ambiguous: Set[str] = set() for p, t in trees.items(): for n in t.body: if isinstance(n, ast.ClassDef): s = _init_sig(n) if s is not None: s.location = (str(p), n.lineno) s.bases = [b.id for b in n.bases if isinstance(b, ast.Name)] if n.name in class_sigs: ambiguous.add(n.name) else: class_sigs[n.name] = s elif isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)): s = _sig_from_args(n.args) s.location = (str(p), n.lineno) if n.name in func_sigs: ambiguous.add(n.name) else: func_sigs[n.name] = s for name, s in class_sigs.items(): for base in s.bases: if base in class_sigs and base != name and base not in ambiguous: s.params |= class_sigs[base].params s.has_var_kw = s.has_var_kw or class_sigs[base].has_var_kw if "R1" in wanted: _check_r1(trees, class_sigs, func_sigs, ambiguous, findings) if "R2" in wanted: _check_r2(trees, files, findings) if "R3" in wanted: _check_r3(trees, findings) if "R4" in wanted: _check_r4(trees, findings) findings.sort(key=lambda f: (f.file, f.line, f.rule)) return findings, notes # ---------------------------------------------------------------- R1 def _check_r1(trees, class_sigs, func_sigs, ambiguous, findings) -> None: for p, t in trees.items(): for node in ast.walk(t): if not isinstance(node, ast.Call): continue target = None if isinstance(node.func, ast.Name): target = node.func.id elif isinstance(node.func, ast.Attribute): target = node.func.attr if not target or target in ambiguous: continue sig = class_sigs.get(target) or func_sigs.get(target) if sig is None: continue for kw in node.keywords: if kw.arg is None: continue if kw.arg not in sig.params and not sig.has_var_kw: where = f"{sig.location[0]}:{sig.location[1]}" if sig.location else "unknown location" findings.append(Finding( "unexpected_kwarg", "error", str(p), node.lineno, f"{target}() called with unexpected keyword '{kw.arg}' " f"(callee at {where} does not accept it; partial patch apply?)", )) # ---------------------------------------------------------------- R2 def _check_r2(trees, files, findings) -> None: defined_locs: Dict[str, List[Tuple[str, int]]] = {} read_locs: Dict[str, List[Tuple[str, int]]] = {} env_keys: Set[str] = set() for p in files: if p.name == ".env": for line in p.read_text(encoding="utf-8", errors="replace").splitlines(): m = ENV_KEY_RE.match(line) if m: env_keys.add(m.group(1)) continue t = trees.get(p) if t is None: continue for node in ast.walk(t): if isinstance(node, ast.Assign): if isinstance(node.value, ast.Dict): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id.lower() in CONFIG_NAMES: for k in node.value.keys: if isinstance(k, ast.Constant) and isinstance(k.value, str): defined_locs.setdefault(k.value, []).append((str(p), node.lineno)) for tgt in node.targets: if (isinstance(tgt, ast.Subscript) and isinstance(tgt.slice, ast.Constant) and isinstance(tgt.slice.value, str)): recv = _recv_name(tgt.value) if recv and recv.lower() in CONFIG_NAMES: defined_locs.setdefault(tgt.slice.value, []).append((str(p), node.lineno)) elif isinstance(node, ast.AnnAssign) and isinstance(node.value, ast.Dict): if isinstance(node.target, ast.Name) and node.target.id.lower() in CONFIG_NAMES: for k in node.value.keys: if isinstance(k, ast.Constant) and isinstance(k.value, str): defined_locs.setdefault(k.value, []).append((str(p), node.lineno)) elif isinstance(node, ast.Call): f = node.func if isinstance(f, ast.Attribute) and f.attr in ("get", "getenv"): recv = _recv_name(f.value) if f.attr == "get" else None if f.attr == "getenv" and recv == "os": recv = "environ" if recv and (recv.lower() in CONFIG_NAMES or recv == "environ"): if node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str): read_locs.setdefault(node.args[0].value, []).append((str(p), node.lineno)) elif isinstance(f, ast.Name) and f.id == "getenv": if node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str): read_locs.setdefault(node.args[0].value, []).append((str(p), node.lineno)) elif isinstance(node, ast.Subscript): if isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str): recv = _recv_name(node.value) if recv and (recv.lower() in CONFIG_NAMES or recv == "environ"): read_locs.setdefault(node.slice.value, []).append((str(p), node.lineno)) if not defined_locs and not env_keys: return # no config source to drift from for k in sorted(set(read_locs) - set(defined_locs) - env_keys): p, ln = read_locs[k][0] findings.append(Finding( "config_drift", "error", p, ln, f"config key '{k}' is read but never defined anywhere. It will silently default.", )) for k in sorted(set(defined_locs) - set(read_locs)): p, ln = defined_locs[k][0] findings.append(Finding( "config_drift", "warning", p, ln, f"config key '{k}' is defined but never read. Dead config.", )) # ---------------------------------------------------------------- R3 def _check_r3(trees, findings) -> None: counts: Dict[str, List[Tuple[str, int]]] = {} for p, t in trees.items(): for node, parent in _walk_with_parent(t): if not isinstance(node, ast.Constant): continue v = node.value if isinstance(v, bool) or not isinstance(v, (int, float)): continue if parent is None: continue if isinstance(parent, ast.Subscript) and parent.slice is node: continue if isinstance(parent, ast.Dict) and node in parent.keys: continue if isinstance(parent, (ast.Assign, ast.AnnAssign)): targets = parent.targets if isinstance(parent, ast.Assign) else [parent.target] if any(isinstance(tgt, ast.Name) and UPPER_NAME_RE.match(tgt.id) for tgt in targets): continue key = repr(float(v)) if key in MAGIC_EXCLUDE: continue counts.setdefault(key, []).append((str(p), node.lineno)) for key, locs in sorted(counts.items()): if len(locs) >= MAGIC_MIN_COUNT: samples = ", ".join(f"{f}:{l}" for f, l in locs[:3]) findings.append(Finding( "magic_number", "warning", locs[0][0], locs[0][1], f"{key} appears {len(locs)} times ({samples}). Hardcoded value doing a named constant's job.", )) # ---------------------------------------------------------------- R4 def _check_r4(trees, findings) -> None: for p, t in trees.items(): defined: Set[str] = set() loads: Dict[str, List[int]] = {} skip_file = False for node in ast.walk(t): if isinstance(node, (ast.Assign, ast.AnnAssign)): targets = node.targets if isinstance(node, ast.Assign) else [node.target] for tgt in targets: _add_target_names(tgt, defined) elif isinstance(node, ast.For): _add_target_names(node.target, defined) elif isinstance(node, ast.With): for item in node.items: if item.optional_vars is not None: _add_target_names(item.optional_vars, defined) elif isinstance(node, ast.ExceptHandler): if node.name: defined.add(node.name) elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): defined.add(node.name) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): a = node.args for grp in (a.posonlyargs, a.args, a.kwonlyargs): for x in grp: defined.add(x.arg) if a.vararg: defined.add(a.vararg.arg) if a.kwarg: defined.add(a.kwarg.arg) elif isinstance(node, ast.Lambda): a = node.args for grp in (a.posonlyargs, a.args, a.kwonlyargs): for x in grp: defined.add(x.arg) if a.vararg: defined.add(a.vararg.arg) if a.kwarg: defined.add(a.kwarg.arg) elif isinstance(node, ast.Import): for alias in node.names: defined.add(alias.asname or alias.name.split(".")[0]) elif isinstance(node, ast.ImportFrom): for alias in node.names: if alias.name == "*": skip_file = True defined.add(alias.asname or alias.name) elif isinstance(node, ast.comprehension): _add_target_names(node.target, defined) elif isinstance(node, ast.NamedExpr): _add_target_names(node.target, defined) if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load): loads.setdefault(node.id, []).append(node.lineno) if skip_file: continue for name, lines in loads.items(): if name in defined or name in BUILTINS: continue if name.startswith("__") and name.endswith("__"): continue findings.append(Finding( "phantom_name", "error", str(p), lines[0], f"'{name}' is used but never defined in this file " f"(lines {', '.join(str(x) for x in lines[:4])}). A typo like this silently defaults.", )) # ---------------------------------------------------------------- CLI def main(argv=None) -> int: ap = argparse.ArgumentParser( prog="drift", description="Find where your code and your config have drifted apart.", ) ap.add_argument("paths", nargs="+", metavar="PATH", help="Python files or directories to scan") ap.add_argument("--rules", default="R1,R2,R3,R4", help="comma-separated rules to run (default: all)") ap.add_argument("--json", action="store_true", help="emit findings as JSON") ap.add_argument("--strict", action="store_true", help="treat warnings as failures too") ap.add_argument("--quiet", action="store_true", help="only print findings, no summary") ap.add_argument("--version", action="version", version=f"drift {VERSION}") args = ap.parse_args(argv) findings, notes = analyze(args.paths, rules=args.rules) errors = [f for f in findings if f.severity == "error"] warnings = [f for f in findings if f.severity == "warning"] failed = bool(errors) or (args.strict and bool(warnings)) if args.json: print(json.dumps([f.to_dict() for f in findings], indent=2)) else: for f in findings: print(f"[{f.severity}] {f.rule} {f.file}:{f.line}") print(f" {f.message}") if not args.quiet: print(f"\ndrift {VERSION}: {len(findings)} finding(s) " f"({len(errors)} error(s), {len(warnings)} warning(s))") for n in notes: print(n, file=sys.stderr) return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())