From 6d18f0f5e731da8bc33d6c4e1c42de78bc9d908b Mon Sep 17 00:00:00 2001 From: Joshua Wink Date: Tue, 13 Oct 2026 08:39:06 -0500 Subject: [PATCH] refactor: the Python code passes its own linter in strict mode The 120 CUP-I008 warnings are resolved. 52 in the code itself: the helper functions of the compiler Filters and of three visitors are now methods of their class, the export pipeline is a named Pipeline, and three small decisions are tables. 68 with a comment that gives the reason, on code that is plain by design: a primitive itself, a visitor of a syntax tree, the entry of a program, a test harness. 13 more comments are on the conformance judge and speed check, which the repository self-check also covers. CI runs the invariant linter with --strict. The self-check test pins the number of comments, so a new one is a change that someone sees. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 2 +- conformance/bench.py | 5 + conformance/run.py | 8 + conformance/runners/python/runner.py | 15 + .../anonymous_steps/__main__.py | 1 + .../apply_suppressions_filter.py | 26 +- .../codeupipe_lint/anonymous_steps/finding.py | 8 +- .../anonymous_steps/lint_pipeline.py | 6 +- .../anonymous_steps/python_step_visitor.py | 81 +-- .../codeupipe_lint/anonymous_steps/rules.py | 21 +- .../codeupipe_lint/invariants/__main__.py | 1 + .../invariants/lint_pipeline.py | 9 +- .../invariants/outside_write_visitor.py | 2 + .../invariants/payload_mutation_visitor.py | 19 +- .../invariants/primitive_kind.py | 2 + .../codeupipe_lint/invariants/rules.py | 3 +- .../invariants/state_read_visitor.py | 34 +- .../codeupipe/codeupipe/compile/__main__.py | 2 + .../filters/emit_inline_module_filter.py | 369 +++++------ .../filters/emit_orchestrator_filter.py | 254 +++---- .../filters/emit_plain_files_filter.py | 56 +- .../compile/filters/load_pipeline_filter.py | 96 +-- .../filters/parse_filter_source_filter.py | 621 +++++++++--------- .../filters/rewrite_payload_ops_filter.py | 33 +- packages/codeupipe/codeupipe/compile/ir.py | 2 +- .../compile/pipelines/export_pipeline.py | 32 +- .../codeupipe/compile/resolve_target.py | 1 + packages/codeupipe/codeupipe/config.py | 3 + packages/codeupipe/codeupipe/core/payload.py | 1 + packages/codeupipe/codeupipe/core/pipeline.py | 1 + .../codeupipe/codeupipe/core/step_label.py | 1 + scorecard/case_gen.py | 6 + scorecard/compare.py | 4 + scorecard/differential.py | 6 + scorecard/minimize.py | 4 + scorecard/modes.py | 4 + scorecard/snapshot_hook.py | 11 +- scorecard/strict.py | 1 + scorecard/vocab.py | 4 + tests/lint/invariants/test_repo_is_clean.py | 14 +- 40 files changed, 931 insertions(+), 838 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a4fa748..f0efbda 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,7 +34,7 @@ jobs: - name: Lint, anonymous steps run: python -m codeupipe_lint.anonymous_steps - name: Lint, invariants - run: python -m codeupipe_lint.invariants + run: python -m codeupipe_lint.invariants --strict - name: Conformance run: python conformance/run.py diff --git a/conformance/bench.py b/conformance/bench.py index 52f0318..575ef6d 100644 --- a/conformance/bench.py +++ b/conformance/bench.py @@ -57,10 +57,12 @@ def python_key(): return "%d.%d" % sys.version_info[:2] +# cup: allow-plain the speed check of the conformance suite: it times the scenarios and compares them with the baseline def load_baseline(): return json.loads(BASELINE.read_text()) if BASELINE.exists() else {} +# cup: allow-plain the speed check of the conformance suite: it times the scenarios and compares them with the baseline def compare(baseline, current, tolerance=DEFAULT_TOLERANCE): """Every number that is not within ``tolerance`` percent of its baseline. @@ -84,6 +86,7 @@ def compare(baseline, current, tolerance=DEFAULT_TOLERANCE): return problems +# cup: allow-plain the speed check of the conformance suite: it times the scenarios and compares them with the baseline def _report(problems, tolerance): if not problems: return "Speed check: every number is within %g%% of its baseline." % tolerance @@ -101,6 +104,7 @@ def _report(problems, tolerance): return "\n".join(lines) +# cup: allow-plain the speed check of the conformance suite: it times the scenarios and compares them with the baseline def main(argv): argv = list(argv) record, check = "--record" in argv, "--check" in argv @@ -167,6 +171,7 @@ def main(argv): return 0 +# cup: allow-plain the speed check of the conformance suite: it times the scenarios and compares them with the baseline def markdown(scenarios, table, problems, refused=()): targets = list(table) lines = ["µs per run (best of 5 × 20,000, after 20,000 warmup runs); lower is better.", "", diff --git a/conformance/run.py b/conformance/run.py index 8fab413..ed10c61 100644 --- a/conformance/run.py +++ b/conformance/run.py @@ -51,6 +51,7 @@ def _typescript(target): } +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def _node_runs_typescript(): """Whether there is a Node that runs a .ts file as it is (22.18 or later).""" node = shutil.which("node") @@ -83,6 +84,7 @@ def _node_runs_typescript(): } +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def default_targets(): """Every target that this machine can run. TypeScript needs Node, and the static TypeScript target needs its package to be installed.""" @@ -93,6 +95,7 @@ def can_run(name): return [name for name in TARGETS if can_run(name)] +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def load_cases(): cases = [] for path in sorted(CASES.glob("*.json")): @@ -104,6 +107,7 @@ def load_cases(): return cases +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def run_target(name, spec, cases): env = dict(os.environ) env.update(spec["env"]) @@ -119,6 +123,7 @@ def run_target(name, spec, cases): return None, f"unreadable runner output ({error}): {proc.stdout[:500]}" +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def _norm(value): """JSON equality: 1 == 1.0, but true is not 1.""" if isinstance(value, bool) or value is None or isinstance(value, str): @@ -132,6 +137,7 @@ def _norm(value): return value +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def judge(case, result): """Return (status, problems).""" if result is None: @@ -179,6 +185,7 @@ def judge(case, result): return "pass", [] +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def main(argv): cases = load_cases() names = argv or default_targets() @@ -211,6 +218,7 @@ def main(argv): return 1 if failed else 0 +# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def _markdown(cases, names, matrix, report): mark = {"pass": "✅", "partial": "✅*", "FAIL": "❌", "unsupported": "➖", "—": "—"} lines = ["| case | " + " | ".join(names) + " |", "|---|" + "---|" * len(names)] diff --git a/conformance/runners/python/runner.py b/conformance/runners/python/runner.py index 52039ec..d450a58 100644 --- a/conformance/runners/python/runner.py +++ b/conformance/runners/python/runner.py @@ -36,6 +36,7 @@ class Unsupported(Exception): pass +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _filter(spec, run_async=False): op = spec["op"] if op == "run_again": @@ -61,6 +62,7 @@ def _filter(spec, run_async=False): raise Unsupported(f"unknown filter op {op!r}") +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _tap(spec): if spec["op"] == "record": return ops.RecordTap(spec["key"]) @@ -71,6 +73,7 @@ def _tap(spec): raise Unsupported(f"unknown tap op {spec['op']!r}") +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _predicate_source(name, when): """Valve predicates are generated as real predicate classes so static lowering can recover their source (it refuses anything else).""" @@ -87,6 +90,7 @@ def _predicate_source(name, when): f"when_{name} = When_{name}()\n") +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _predicates(steps, module_dir, module_name): sources = [] @@ -107,6 +111,7 @@ def walk(items): sys.path.remove(str(module_dir)) +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def build(case, cls, predicates, names, run_async=False): p = cls() for step in case["steps"]: @@ -147,6 +152,7 @@ def _named(steps, label): return [label.get(step, step) for step in steps] +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _named_hooks(events, label): out = [] for event in events: @@ -155,6 +161,7 @@ def _named_hooks(events, label): return out +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _dynamic(case, predicates, run_async=False): names = {} p = build(case, Pipeline, predicates, names, run_async) @@ -197,6 +204,7 @@ def report(state): return result +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _load_export(p, lowering, observe, workdir): """Export ``p`` and import the package from disk, as a customer would.""" from codeupipe.compile import build_export_pipeline_v2 @@ -217,6 +225,7 @@ def _load_export(p, lowering, observe, workdir): sys.path.remove(package_parent) +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _static(case, predicates, lowering, workdir, observe=False): # The compiled form always gets its State from the caller, so "given_state" needs nothing more. names = {} @@ -263,6 +272,7 @@ def report(one): return result +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def run_case(case, target, index): sink.reset() tmp = pathlib.Path(tempfile.mkdtemp(prefix="cup_conf_")) @@ -300,6 +310,7 @@ def _reference_write(data, key): return data +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _reference(): """The reference workload: ten dict writes behind ten function calls, with no framework. A pipeline's time is reported relative to it.""" @@ -309,6 +320,7 @@ def _reference(): return data +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _timed(fn): import time t0 = time.perf_counter() @@ -317,6 +329,7 @@ def _timed(fn): return time.perf_counter() - t0 +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def _measure(fn): """(µs per run, time relative to the reference workload). @@ -339,6 +352,7 @@ def _measure(fn): return best / BENCH_RUNS * 1e6, statistics.median(ratios) +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def bench(target, scenarios): results = [] for index, case in enumerate(scenarios): @@ -372,6 +386,7 @@ def bench(target, scenarios): return results +# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def main(): target = sys.argv[1] if target.startswith("bench:"): diff --git a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/__main__.py b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/__main__.py index 956f743..28330ec 100644 --- a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/__main__.py +++ b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/__main__.py @@ -16,6 +16,7 @@ from ..rule_ids import unknown_rules +# cup: allow-plain the entry of the program: it reads the arguments, runs the pipeline and prints the result def main(argv=None) -> int: parser = argparse.ArgumentParser(prog="python -m codeupipe_lint.anonymous_steps", description="Flag anonymous pipeline steps (packages/codeupipe-lint/README.md).") diff --git a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/apply_suppressions_filter.py b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/apply_suppressions_filter.py index d406663..bfa4d17 100644 --- a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/apply_suppressions_filter.py +++ b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/apply_suppressions_filter.py @@ -37,25 +37,25 @@ def call(self, payload: Payload) -> Payload: except OSError: lines = [] cache[finding.path] = lines - places = _marker_lines(finding, lines, self.marker) + places = ApplySuppressionsFilter._marker_lines(finding, lines, self.marker) used.update((finding.path, place) for place in places) (suppressed if places else kept).append(finding) return (payload.insert("findings", kept).insert("suppressed", suppressed) .insert("used_markers", sorted(used))) + @staticmethod + def _line(lines, number): + return lines[number - 1] if number and 0 < number <= len(lines) else "" -def _line(lines, number): - return lines[number - 1] if number and 0 < number <= len(lines) else "" - - -def _marker_lines(finding, lines, marker=MARKER): - """The lines of the comments that allow this finding. Empty when none does.""" - places = [number for number in (finding.line, finding.call_line) if marker.search(_line(lines, number))] - above = _line(lines, finding.line - 1) - if _COMMENT_ONLY.match(above) and marker.search(above): - places.append(finding.line - 1) - return places + @staticmethod + def _marker_lines(finding, lines, marker=MARKER): + """The lines of the comments that allow this finding. Empty when none does.""" + places = [number for number in (finding.line, finding.call_line) if marker.search(ApplySuppressionsFilter._line(lines, number))] + above = ApplySuppressionsFilter._line(lines, finding.line - 1) + if _COMMENT_ONLY.match(above) and marker.search(above): + places.append(finding.line - 1) + return places def _is_suppressed(finding, lines, marker=MARKER) -> bool: - return bool(_marker_lines(finding, lines, marker)) + return bool(ApplySuppressionsFilter._marker_lines(finding, lines, marker)) diff --git a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/finding.py b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/finding.py index db0eb29..6f0d32d 100644 --- a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/finding.py +++ b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/finding.py @@ -3,6 +3,10 @@ from typing import NamedTuple, Optional +# What a report line has in front of the message, for each severity. +_PREFIX = {"warning": "warning: "} + + class Finding(NamedTuple): """An anonymous step at ``path:line:col`` (1-based line and column). @@ -23,5 +27,5 @@ class Finding(NamedTuple): severity: Optional[str] = None def format(self) -> str: - warning = "warning: " if self.severity == "warning" else "" - return "%s:%d:%d %s %s %s%s" % (self.path, self.line, self.col, self.rule, self.slug, warning, self.message) + prefix = _PREFIX.get(self.severity, "") + return "%s:%d:%d %s %s %s%s" % (self.path, self.line, self.col, self.rule, self.slug, prefix, self.message) diff --git a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/lint_pipeline.py b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/lint_pipeline.py index ae441ac..f8e0b7d 100644 --- a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/lint_pipeline.py +++ b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/lint_pipeline.py @@ -34,8 +34,6 @@ def lint_paths(paths: Iterable[str], exclude: Optional[Iterable[str]] = None, """Run the linter over ``paths``; the result Payload holds ``findings``, ``suppressed``, ``errors``, ``report``, ``error_report`` and ``exit_code``.""" data = {"paths": list(paths), "findings": [], "errors": []} - if exclude: - data["exclude"] = list(exclude) - if disable: - data["disable"] = list(disable) + data["exclude"] = list(exclude or ()) + data["disable"] = list(disable or ()) return build_lint_pipeline().run(Payload(data)) diff --git a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/python_step_visitor.py b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/python_step_visitor.py index 074f7fe..40fee67 100644 --- a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/python_step_visitor.py +++ b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/python_step_visitor.py @@ -45,39 +45,7 @@ _FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef) -def _is_plain_pipeline_call(node: ast.AST) -> bool: - if not isinstance(node, ast.Call) or node.args or node.keywords: - return False - func = node.func - return (isinstance(func, ast.Name) and func.id == "Pipeline") or ( - isinstance(func, ast.Attribute) and func.attr == "Pipeline") - - -def _own_nodes(body: List[ast.stmt]): - """Every node in ``body``, not descending into nested functions, classes or lambdas.""" - stack = list(reversed(body)) - while stack: - node = stack.pop() - yield node - for child in reversed(list(ast.iter_child_nodes(node))): - if isinstance(child, _FUNCS + (ast.ClassDef, ast.Lambda)): - yield child # the definition itself binds a name; its body is its own scope - continue - stack.append(child) - - -def _target_names(target: ast.AST): - if isinstance(target, ast.Name): - yield target.id - elif isinstance(target, (ast.Tuple, ast.List)): - for element in target.elts: - for name in _target_names(element): - yield name - elif isinstance(target, ast.Starred): - for name in _target_names(target.value): - yield name - - +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class PythonStepVisitor: """Collects Findings for one module (see the module docstring).""" @@ -104,7 +72,7 @@ def bind(name: str, binding: Binding) -> None: for arg in (args.vararg, args.kwarg): if arg is not None: bind(arg.arg, _OTHER) - nodes = list(_own_nodes(body)) + nodes = list(PythonStepVisitor._own_nodes(body)) local_classes = set() for node in nodes: if isinstance(node, _FUNCS): @@ -118,16 +86,16 @@ def bind(name: str, binding: Binding) -> None: targets = node.targets if isinstance(node, ast.Assign) else [node.target] binding = self._value_binding(node.value, local_classes, owner) for target in targets: - names = list(_target_names(target)) + names = list(PythonStepVisitor._target_names(target)) for name in names: bind(name, binding if isinstance(target, ast.Name) else _OTHER) elif isinstance(node, (ast.AugAssign, ast.For, ast.AsyncFor, ast.comprehension, ast.NamedExpr)): - for name in _target_names(node.target): + for name in PythonStepVisitor._target_names(node.target): bind(name, _OTHER) elif isinstance(node, (ast.With, ast.AsyncWith)): for item in node.items: if item.optional_vars is not None: - for name in _target_names(item.optional_vars): + for name in PythonStepVisitor._target_names(item.optional_vars): bind(name, _OTHER) elif isinstance(node, (ast.Import, ast.ImportFrom)): for alias in node.names: @@ -148,7 +116,7 @@ def bind(name: str, binding: Binding) -> None: def _value_binding(value: Optional[ast.AST], local_classes: Set[str], owner: Optional[str]) -> Binding: if isinstance(value, ast.Lambda): return ("lambda", "", owner, value) - if _is_plain_pipeline_call(value): + if PythonStepVisitor._is_plain_pipeline_call(value): return ("plain_pipeline", "", owner, value) if isinstance(value, ast.Call) and isinstance(value.func, ast.Name) and value.func.id in local_classes: return ("local_instance", value.func.id, owner, value) @@ -188,7 +156,7 @@ def _describe(self, expr: ast.AST, chain: List[Dict[str, Binding]], if isinstance(expr, ast.Name): got = self._describe_binding(self._resolve(expr.id, chain)) return None if got is None else (got[0], expr.id, got[1]) - if _is_plain_pipeline_call(expr): + if PythonStepVisitor._is_plain_pipeline_call(expr): return UNNAMED_PIPELINE, None, "a plain Pipeline()" if isinstance(expr, ast.Call) and isinstance(expr.func, ast.Name): name = expr.func.id @@ -227,7 +195,7 @@ def _returned(self, func: ast.AST, parent_chain: List[Dict[str, Binding]], return None visiting = visiting | {id(func)} chain = self._chain_for(func, parent_chain) - for node in _own_nodes(func.body): + for node in PythonStepVisitor._own_nodes(func.body): if isinstance(node, ast.Return) and node.value is not None: got = self._describe(node.value, chain, visiting) if got is not None: @@ -278,3 +246,36 @@ def _report(self, expr: ast.AST, role: str, call: ast.Call, chain: List[Dict[str self._seen.add(key) self._findings.append(make_finding("python", self._path, expr.lineno, expr.col_offset + 1, rule, role, what, call.lineno)) + + @staticmethod + def _is_plain_pipeline_call(node: ast.AST) -> bool: + if not isinstance(node, ast.Call) or node.args or node.keywords: + return False + func = node.func + return (isinstance(func, ast.Name) and func.id == "Pipeline") or ( + isinstance(func, ast.Attribute) and func.attr == "Pipeline") + + @staticmethod + def _own_nodes(body: List[ast.stmt]): + """Every node in ``body``, not descending into nested functions, classes or lambdas.""" + stack = list(reversed(body)) + while stack: + node = stack.pop() + yield node + for child in reversed(list(ast.iter_child_nodes(node))): + if isinstance(child, _FUNCS + (ast.ClassDef, ast.Lambda)): + yield child # the definition itself binds a name; its body is its own scope + continue + stack.append(child) + + @staticmethod + def _target_names(target: ast.AST): + if isinstance(target, ast.Name): + yield target.id + elif isinstance(target, (ast.Tuple, ast.List)): + for element in target.elts: + for name in PythonStepVisitor._target_names(element): + yield name + elif isinstance(target, ast.Starred): + for name in PythonStepVisitor._target_names(target.value): + yield name diff --git a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/rules.py b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/rules.py index 759aa4c..84abb75 100644 --- a/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/rules.py +++ b/packages/codeupipe-lint/codeupipe_lint/anonymous_steps/rules.py @@ -40,18 +40,21 @@ } +# The rules that have a fix of their own. {pipeline} is the example of a named pipeline in the language. +_FIX_OF_RULE = { + UNNAMED_PIPELINE: ("A nested pipeline must be a named pipeline type ({pipeline}), named like a filter. " + "A plain Pipeline can only run at the top level."), + FUNCTION_PREDICATE: ("Valve predicates are objects (conformance/API.md, \"Valve\"): wrap the condition in a " + "named class with test(payload), configuration on the instance, and pass an instance."), + PARSE_ERROR: "Fix the syntax error so the file can be checked.", +} + + def _fix(lang: str, rule: str) -> str: - if rule == UNNAMED_PIPELINE: - return ("A nested pipeline must be a named pipeline type (%s), named like a filter. " - "A plain Pipeline can only run at the top level." % _PIPELINE_EXAMPLE[lang]) - if rule == FUNCTION_PREDICATE: - return ("Valve predicates are objects (conformance/API.md, \"Valve\"): wrap the condition in a " - "named class with test(payload), configuration on the instance, and pass an instance.") - if rule == PARSE_ERROR: - return "Fix the syntax error so the file can be checked." - return _FIX[lang] + return _FIX_OF_RULE.get(rule, _FIX[lang]).replace("{pipeline}", _PIPELINE_EXAMPLE[lang]) +# cup: allow-plain it makes one Finding from its parts, for each analysis Filter def make_finding(lang: str, path: str, line: int, col: int, rule: str, role: str, what: str, call_line: Optional[int] = None) -> Finding: """A Finding whose message reads `` . ``; ``what`` starts with its verb diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/__main__.py b/packages/codeupipe-lint/codeupipe_lint/invariants/__main__.py index 8cdf99a..61db748 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/__main__.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/__main__.py @@ -16,6 +16,7 @@ from ..rule_ids import unknown_rules +# cup: allow-plain the entry of the program: it reads the arguments, runs the pipeline and prints the result def main(argv=None) -> int: parser = argparse.ArgumentParser(prog="python -m codeupipe_lint.invariants", description="Enforce CUP's structural invariants (packages/codeupipe-lint/README.md).") diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/lint_pipeline.py b/packages/codeupipe-lint/codeupipe_lint/invariants/lint_pipeline.py index 00befb0..ee3e1e1 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/lint_pipeline.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/lint_pipeline.py @@ -60,11 +60,6 @@ def lint_paths(paths: Iterable[str], exclude: Optional[Iterable[str]] = None, disable: Optional[Iterable[str]] = None, strict: bool = False) -> Payload: """Run the linter over ``paths``; the result Payload holds ``findings``, ``suppressed``, ``errors``, ``report``, ``error_report`` and ``exit_code``.""" - data = {"paths": list(paths), "findings": [], "errors": []} - if exclude: - data["exclude"] = list(exclude) - if disable: - data["disable"] = list(disable) - if strict: - data["strict"] = True + data = {"paths": list(paths), "findings": [], "errors": [], + "exclude": list(exclude or ()), "disable": list(disable or ()), "strict": bool(strict)} return build_lint_pipeline().run(Payload(data)) diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/outside_write_visitor.py b/packages/codeupipe-lint/codeupipe_lint/invariants/outside_write_visitor.py index 7ebafe3..97df595 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/outside_write_visitor.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/outside_write_visitor.py @@ -19,6 +19,7 @@ WRITES = {"insert", "remove", "merge", "with_trace"} +# cup: allow-plain used by the analysis Filters and the visitor: it finds the payload parameter of a method def payload_parameter(method): """The parameter that holds the Payload: the one named ``payload``, else the last one.""" names = [a.arg for a in method.args.posonlyargs + method.args.args] @@ -29,6 +30,7 @@ def payload_parameter(method): return "payload" if "payload" in names else names[-1] +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class OutsideWriteVisitor(ast.NodeVisitor): def __init__(self, path, method): self.path = path diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/payload_mutation_visitor.py b/packages/codeupipe-lint/codeupipe_lint/invariants/payload_mutation_visitor.py index 47649e7..ec07af5 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/payload_mutation_visitor.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/payload_mutation_visitor.py @@ -27,13 +27,7 @@ "symmetric_difference_update", "appendleft", "extendleft", "popleft"} -def _base(node): - """The name or call an attribute/subscript chain hangs off.""" - while isinstance(node, (ast.Subscript, ast.Attribute)): - node = node.value - return node - - +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class PayloadMutationVisitor(ast.NodeVisitor): def __init__(self, path, method): self.path = path @@ -90,7 +84,7 @@ def _check_write_target(self, target, statement): if self._is_internals(chain): self._report(statement, PAYLOAD_INTERNALS, "Writes a Payload's internals (`_data`)") return - base = _base(target) + base = PayloadMutationVisitor._base(target) if isinstance(target, ast.Subscript) and (self._is_get(base) or (isinstance(base, ast.Name) and base.id in self.shared)): self._report(statement, SHARED_VALUE_MUTATION, self._describe(base)) @@ -146,7 +140,7 @@ def visit_Call(self, node): if self._is_internals(receiver): self._report(node, PAYLOAD_INTERNALS, "Writes a Payload's internals (`_data`)") else: - base = _base(receiver) + base = PayloadMutationVisitor._base(receiver) if self._is_get(base) or (isinstance(base, ast.Name) and base.id in self.shared): self._report(node, SHARED_VALUE_MUTATION, self._describe(base)) self.generic_visit(node) @@ -155,3 +149,10 @@ def _skip(self, node): """A nested function or class is its own scope; step methods are found from the module.""" visit_FunctionDef = visit_AsyncFunctionDef = visit_ClassDef = visit_Lambda = _skip + + @staticmethod + def _base(node): + """The name or call an attribute/subscript chain hangs off.""" + while isinstance(node, (ast.Subscript, ast.Attribute)): + node = node.value + return node diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/primitive_kind.py b/packages/codeupipe-lint/codeupipe_lint/invariants/primitive_kind.py index 602c601..8595861 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/primitive_kind.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/primitive_kind.py @@ -23,6 +23,7 @@ _HOOK_BASES = {"Hook", "AsyncHook"} +# cup: allow-plain shared by the analysis Filters: it says which primitive a class is, from its shape def _base_names(node): return {base.id if isinstance(base, ast.Name) else getattr(base, "attr", "") for base in node.bases} @@ -33,6 +34,7 @@ def _takes_only_the_payload(method) -> bool: and not method.args.vararg and not method.args.kwonlyargs and not method.args.kwarg) +# cup: allow-plain shared by the analysis Filters: it says which primitive a class is, from its shape def primitive_kind(node): """"Pipeline", "Hook", "Filter", "Tap" or "predicate"; None for any other class.""" methods = {member.name: member for member in node.body diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/rules.py b/packages/codeupipe-lint/codeupipe_lint/invariants/rules.py index 8fc085c..2adbfe7 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/rules.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/rules.py @@ -19,6 +19,7 @@ # The rules whose findings are warnings. WARNINGS = frozenset([LOGIC_OUTSIDE_PRIMITIVE, MARKER_ALLOWS_NOTHING]) +_SEVERITY = {rule: "warning" for rule in WARNINGS} RULES = { PARSE_ERROR: "parse-error", @@ -60,4 +61,4 @@ def make_finding(path: str, line: int, col: int, rule: str, what: str) -> Finding: """A Finding whose message reads ``. ``.""" return Finding(path, line, col, rule, RULES[rule], "%s. %s" % (what, _FIX[rule]), "python", - severity="warning" if rule in WARNINGS else None) + severity=_SEVERITY.get(rule)) diff --git a/packages/codeupipe-lint/codeupipe_lint/invariants/state_read_visitor.py b/packages/codeupipe-lint/codeupipe_lint/invariants/state_read_visitor.py index f24c811..e4fd210 100644 --- a/packages/codeupipe-lint/codeupipe_lint/invariants/state_read_visitor.py +++ b/packages/codeupipe-lint/codeupipe_lint/invariants/state_read_visitor.py @@ -24,23 +24,11 @@ "record_timing", "record_error"} -def _is_state_type(node) -> bool: - if isinstance(node, ast.Name): - return node.id == "State" - if isinstance(node, ast.Attribute): - return node.attr == "State" - return isinstance(node, ast.Constant) and node.value == "State" - - def _is_self(node) -> bool: return isinstance(node, ast.Name) and node.id == "self" -def _names_a_pipeline(node) -> bool: - name = node.id if isinstance(node, ast.Name) else node.attr if isinstance(node, ast.Attribute) else "" - return "pipeline" in name.lower() - - +# cup: allow-plain used by the analysis Filter and the visitor: it finds the names that hold a State def state_handles(class_node) -> set: """Attributes that ``__init__`` sets to a State.""" handles = set() @@ -48,19 +36,20 @@ def state_handles(class_node) -> set: if not (isinstance(member, ast.FunctionDef) and member.name == "__init__"): continue state_parameters = {a.arg for a in member.args.args + member.args.kwonlyargs - if a.annotation is not None and _is_state_type(a.annotation)} + if a.annotation is not None and StateReadVisitor._is_state_type(a.annotation)} for node in ast.walk(member): if not isinstance(node, ast.Assign): continue value = node.value is_state = ((isinstance(value, ast.Name) and value.id in state_parameters) or (isinstance(value, ast.Attribute) and value.attr == "state") - or (isinstance(value, ast.Call) and _is_state_type(value.func))) + or (isinstance(value, ast.Call) and StateReadVisitor._is_state_type(value.func))) if is_state: handles.update(t.attr for t in node.targets if isinstance(t, ast.Attribute) and _is_self(t.value)) return handles +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class StateReadVisitor(ast.NodeVisitor): def __init__(self, path, method, handles): self.path = path @@ -75,7 +64,7 @@ def _reaches_state(self, node) -> bool: if _is_self(node.value) and node.attr in self.handles: return True if node.attr == "state" and not _is_self(node.value): - if outer in STATE_MEMBERS or _names_a_pipeline(node.value): + if outer in STATE_MEMBERS or StateReadVisitor._names_a_pipeline(node.value): return True outer = node.attr node = node.value @@ -92,3 +81,16 @@ def _skip(self, node): """A nested function or class is its own scope.""" visit_FunctionDef = visit_AsyncFunctionDef = visit_ClassDef = visit_Lambda = _skip + + @staticmethod + def _is_state_type(node) -> bool: + if isinstance(node, ast.Name): + return node.id == "State" + if isinstance(node, ast.Attribute): + return node.attr == "State" + return isinstance(node, ast.Constant) and node.value == "State" + + @staticmethod + def _names_a_pipeline(node) -> bool: + name = node.id if isinstance(node, ast.Name) else node.attr if isinstance(node, ast.Attribute) else "" + return "pipeline" in name.lower() diff --git a/packages/codeupipe/codeupipe/compile/__main__.py b/packages/codeupipe/codeupipe/compile/__main__.py index eb23c53..e6082fd 100644 --- a/packages/codeupipe/codeupipe/compile/__main__.py +++ b/packages/codeupipe/codeupipe/compile/__main__.py @@ -26,6 +26,7 @@ def _usage_error(message: str) -> int: return 2 +# cup: allow-plain the entry of the program: it reads the arguments, runs the pipeline and prints the result def export(target: str, out_dir: str, lowering: str = "steps", observe: bool = False, overwrite: bool = False) -> int: """Export one pipeline; the exit code for it.""" if observe and lowering == "inline": @@ -58,6 +59,7 @@ def export(target: str, out_dir: str, lowering: str = "steps", observe: bool = F return 0 +# cup: allow-plain the entry of the program: it reads the arguments, runs the pipeline and prints the result def main(argv=None) -> int: parser = argparse.ArgumentParser(prog="python -m codeupipe.compile", description="Export a Pipeline to plain Python (docs/compile.md).") diff --git a/packages/codeupipe/codeupipe/compile/filters/emit_inline_module_filter.py b/packages/codeupipe/codeupipe/compile/filters/emit_inline_module_filter.py index f21f4fa..d6d3c93 100644 --- a/packages/codeupipe/codeupipe/compile/filters/emit_inline_module_filter.py +++ b/packages/codeupipe/codeupipe/compile/filters/emit_inline_module_filter.py @@ -53,19 +53,19 @@ def call(self, payload: Payload) -> Payload: for step in pipeline_ir.steps: prefix = f"_{step.ident}__" - renames = _definition_renames(step, prefix) + renames = EmitInlineModuleFilter._definition_renames(step, prefix) helpers: Dict[str, ast.FunctionDef] = {} for line in step.imports: - _merge_import(imports, line, step) + EmitInlineModuleFilter._merge_import(imports, line, step) for source in step.definitions: tree = ast.parse(source) - _rename_names(tree, renames) + EmitInlineModuleFilter._rename_names(tree, renames) helpers.update({n.name: n for n in tree.body if isinstance(n, ast.FunctionDef)}) definitions.append(ast.unparse(tree)) call_ast = _collapse_double_dict(_renamed_copy(step.call_ast, renames)) if step.kind == "tap": - body.extend(_tap_lines(step, call_ast, prefix, functions)) + body.extend(EmitInlineModuleFilter._tap_lines(step, call_ast, prefix, functions)) elif step.kind == "valve": predicate = (_collapse_double_dict(_renamed_copy(step.predicate_ast, renames)) if step.predicate_ast else None) @@ -74,12 +74,12 @@ def call(self, payload: Payload) -> Payload: filter_name=step.name, method_name="predicate", line=0, col=0, node_kind="FunctionDef", suggestion="Valve reached the emitter without a predicate.", ) - condition = _predicate_condition(predicate, prefix, functions, helpers) - inner = _filter_lines(step, call_ast, prefix, functions) + condition = EmitInlineModuleFilter._predicate_condition(predicate, prefix, functions, helpers) + inner = EmitInlineModuleFilter._filter_lines(step, call_ast, prefix, functions) body.append(f"if {condition}:") body.extend(_indent(inner or ["pass"])) else: - body.extend(_filter_lines(step, call_ast, prefix, functions)) + body.extend(EmitInlineModuleFilter._filter_lines(step, call_ast, prefix, functions)) body.append("return data") sections = [ @@ -96,58 +96,197 @@ def call(self, payload: Payload) -> Payload: "content": "from .pipeline import run\n\n__all__ = [\"run\"]\n"}) return payload.insert("files", files) + @staticmethod + def _filter_lines(step, fn: ast.FunctionDef, prefix: str, functions: List[str]) -> List[str]: + """Inline a filter body, or call it as a function.""" + if EmitInlineModuleFilter._inlinable(fn, allow_final_return=True): + return EmitInlineModuleFilter._inline_filter(fn, prefix) + functions.append(_as_function(fn, f"{prefix}step")) + return [f"data = {prefix}step(data)"] + + @staticmethod + def _tap_lines(step, fn: ast.FunctionDef, prefix: str, functions: List[str]) -> List[str]: + """Plane B: a Tap failure never halts the run (Exception only).""" + if EmitInlineModuleFilter._inlinable(fn, allow_final_return=False): + inner = [ast.unparse(stmt) for stmt in EmitInlineModuleFilter._body_without_docstring(EmitInlineModuleFilter._rename_locals(fn, prefix))] + else: + functions.append(_as_function(fn, f"{prefix}tap")) + inner = [f"{prefix}tap(data)"] + return ["try:", *_indent(inner or ["pass"]), "except Exception:", " pass"] + + @staticmethod + def _predicate_condition(fn: ast.FunctionDef, prefix: str, functions: List[str], + helpers: Dict[str, ast.FunctionDef]) -> str: + body = EmitInlineModuleFilter._body_without_docstring(fn) + if len(body) == 1 and isinstance(body[0], ast.Return) and body[0].value is not None: + return ast.unparse(EmitInlineModuleFilter._inline_helper_call(body[0].value, helpers)) + functions.append(_as_function(fn, f"{prefix}predicate")) + return f"{prefix}predicate(data)" + + @staticmethod + def _inline_filter(fn: ast.FunctionDef, prefix: str) -> List[str]: + fn = EmitInlineModuleFilter._rename_locals(fn, prefix) + statements = EmitInlineModuleFilter._body_without_docstring(fn) + *head, last = statements + lines = [ast.unparse(stmt) for stmt in head] + value = last.value + if isinstance(value, ast.Name) and value.id == "data": + return lines + return lines + [f"data = {ast.unparse(value)}"] + + @staticmethod + def _inline_helper_call(expr: ast.AST, helpers: Dict[str, ast.FunctionDef]) -> ast.AST: + """``helper(dict(data))`` -> the helper's return expression over ``data``, + when the helper is ``def helper(d): return `` and only reads ``d`` + through ``d.get(...)``. Generated adapters wrap Valve predicates this + way, copying the whole dict on each check.""" + if not (isinstance(expr, ast.Call) and isinstance(expr.func, ast.Name) and len(expr.args) == 1 + and not expr.keywords and _is_dict_call(expr.args[0]) + and isinstance(expr.args[0].args[0], ast.Name) and expr.args[0].args[0].id == "data"): + return expr + helper = helpers.get(expr.func.id) + if helper is None or len(helper.args.args) != 1 or helper.args.vararg or helper.args.kwarg: + return expr + body = EmitInlineModuleFilter._body_without_docstring(helper) + if not (len(body) == 1 and isinstance(body[0], ast.Return) and body[0].value is not None): + return expr + param = helper.args.args[0].arg + returned = copy.deepcopy(body[0].value) + parents = {child: parent for parent in ast.walk(returned) for child in ast.iter_child_nodes(parent)} + for node in ast.walk(returned): + if isinstance(node, ast.Name) and node.id == param: + parent = parents.get(node) + if not (isinstance(parent, ast.Attribute) and parent.attr == "get" + and isinstance(parents.get(parent), ast.Call)): + return expr + node.id = "data" + elif isinstance(node, (ast.Lambda, ast.NamedExpr)): + return expr + return returned + + @staticmethod + def _inlinable(fn: ast.FunctionDef, allow_final_return: bool, any_return: bool = False) -> bool: + body = EmitInlineModuleFilter._body_without_docstring(fn) + if not body: + return not allow_final_return + for node in ast.walk(fn): + if isinstance(node, (ast.Yield, ast.YieldFrom, ast.Await, ast.Global, ast.Nonlocal)): + return False + if isinstance(node, (ast.Lambda, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: + if any(isinstance(n, ast.Name) and n.id == "data" for n in ast.walk(node)): + return False + if isinstance(node, ast.Name) and node.id == "data" and not isinstance(node.ctx, ast.Load): + return False + returns = [n for n in ast.walk(fn) if isinstance(n, ast.Return)] + final = body[-1] if isinstance(body[-1], ast.Return) else None + if any_return: + pass + elif allow_final_return: + if final is None or final.value is None or returns != [final]: + return False + elif returns: + return False + return True + + @staticmethod + def _definition_renames(step, prefix: str) -> Dict[str, str]: + renames: Dict[str, str] = {} + for source in step.definitions: + for stmt in ast.parse(source).body: + for name in EmitInlineModuleFilter._top_level_bound(stmt): + renames[name] = prefix + name + return renames + + @staticmethod + def _top_level_bound(stmt: ast.stmt) -> Set[str]: + if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + return {stmt.name} + names: Set[str] = set() + targets = stmt.targets if isinstance(stmt, ast.Assign) else [getattr(stmt, "target", None)] + for target in targets: + for node in ast.walk(target) if target is not None else (): + if isinstance(node, ast.Name): + names.add(node.id) + return names + + @staticmethod + def _rename_names(tree: ast.AST, renames: Dict[str, str]) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.Name) and node.id in renames: + node.id = renames[node.id] + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name in renames: + node.name = renames[node.name] + + @staticmethod + def _rename_locals(fn: ast.FunctionDef, prefix: str) -> ast.FunctionDef: + """Prefix every name the body binds, so inlined steps cannot collide.""" + fn = copy.deepcopy(fn) + local: Set[str] = set() + for node in ast.walk(fn): + if isinstance(node, ast.Name) and not isinstance(node.ctx, ast.Load): + local.add(node.id) + elif isinstance(node, ast.arg) and node is not fn.args.args[0]: + local.add(node.arg) + elif isinstance(node, ast.ExceptHandler) and node.name: + local.add(node.name) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: + local.add(node.name) + local.discard("data") + renames = {name: prefix + name for name in local} + for node in ast.walk(fn): + if isinstance(node, ast.Name) and node.id in renames: + node.id = renames[node.id] + elif isinstance(node, ast.arg) and node.arg in renames: + node.arg = renames[node.arg] + elif isinstance(node, ast.ExceptHandler) and node.name in renames: + node.name = renames[node.name] + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: + node.name = renames.get(node.name, node.name) + return fn + + @staticmethod + def _body_without_docstring(fn: ast.FunctionDef) -> List[ast.stmt]: + body = list(fn.body) + if body and isinstance(body[0], ast.Expr) and isinstance(getattr(body[0], "value", None), ast.Constant) \ + and isinstance(body[0].value.value, str): + body = body[1:] + return body + + @staticmethod + def _merge_import(imports: Dict[str, str], line: str, step) -> None: + bound = EmitInlineModuleFilter._top_level_bound_import(line) + for name in bound: + existing = imports.get(name) + if existing is not None and existing != line: + raise UnsupportedFilterShape( + filter_name=step.name, method_name="imports", line=0, col=0, node_kind="Import", + suggestion=f"`{name}` is imported differently by two steps ({existing!r} vs {line!r}); " + "use the default per-step lowering for this pipeline.", + ) + imports[name] = line + + @staticmethod + def _top_level_bound_import(line: str) -> Set[str]: + stmt = ast.parse(line).body[0] + if isinstance(stmt, ast.Import): + return {alias.asname or alias.name.split(".")[0] for alias in stmt.names} + return {alias.asname or alias.name for alias in stmt.names} + # -------------------------------------------------------------------------- # steps # -------------------------------------------------------------------------- -def _filter_lines(step, fn: ast.FunctionDef, prefix: str, functions: List[str]) -> List[str]: - """Inline a filter body, or call it as a function.""" - if _inlinable(fn, allow_final_return=True): - return _inline_filter(fn, prefix) - functions.append(_as_function(fn, f"{prefix}step")) - return [f"data = {prefix}step(data)"] - - -def _tap_lines(step, fn: ast.FunctionDef, prefix: str, functions: List[str]) -> List[str]: - """Plane B: a Tap failure never halts the run (Exception only).""" - if _inlinable(fn, allow_final_return=False): - inner = [ast.unparse(stmt) for stmt in _body_without_docstring(_rename_locals(fn, prefix))] - else: - functions.append(_as_function(fn, f"{prefix}tap")) - inner = [f"{prefix}tap(data)"] - return ["try:", *_indent(inner or ["pass"]), "except Exception:", " pass"] - - def _indent(statements: List[str]) -> List[str]: """Indent every line; an unparsed compound statement spans several.""" return [" " + line for stmt in statements for line in stmt.splitlines()] -def _predicate_condition(fn: ast.FunctionDef, prefix: str, functions: List[str], - helpers: Dict[str, ast.FunctionDef]) -> str: - body = _body_without_docstring(fn) - if len(body) == 1 and isinstance(body[0], ast.Return) and body[0].value is not None: - return ast.unparse(_inline_helper_call(body[0].value, helpers)) - functions.append(_as_function(fn, f"{prefix}predicate")) - return f"{prefix}predicate(data)" - - -def _inline_filter(fn: ast.FunctionDef, prefix: str) -> List[str]: - fn = _rename_locals(fn, prefix) - statements = _body_without_docstring(fn) - *head, last = statements - lines = [ast.unparse(stmt) for stmt in head] - value = last.value - if isinstance(value, ast.Name) and value.id == "data": - return lines - return lines + [f"data = {ast.unparse(value)}"] - - # -------------------------------------------------------------------------- # redundant copies # -------------------------------------------------------------------------- +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class _CollapseDoubleDict(ast.NodeTransformer): def visit_Call(self, node: ast.Call): node = self.generic_visit(node) @@ -166,159 +305,21 @@ def _collapse_double_dict(fn: ast.AST) -> ast.AST: return ast.fix_missing_locations(_CollapseDoubleDict().visit(fn)) -def _inline_helper_call(expr: ast.AST, helpers: Dict[str, ast.FunctionDef]) -> ast.AST: - """``helper(dict(data))`` -> the helper's return expression over ``data``, - when the helper is ``def helper(d): return `` and only reads ``d`` - through ``d.get(...)``. Generated adapters wrap Valve predicates this - way, copying the whole dict on each check.""" - if not (isinstance(expr, ast.Call) and isinstance(expr.func, ast.Name) and len(expr.args) == 1 - and not expr.keywords and _is_dict_call(expr.args[0]) - and isinstance(expr.args[0].args[0], ast.Name) and expr.args[0].args[0].id == "data"): - return expr - helper = helpers.get(expr.func.id) - if helper is None or len(helper.args.args) != 1 or helper.args.vararg or helper.args.kwarg: - return expr - body = _body_without_docstring(helper) - if not (len(body) == 1 and isinstance(body[0], ast.Return) and body[0].value is not None): - return expr - param = helper.args.args[0].arg - returned = copy.deepcopy(body[0].value) - parents = {child: parent for parent in ast.walk(returned) for child in ast.iter_child_nodes(parent)} - for node in ast.walk(returned): - if isinstance(node, ast.Name) and node.id == param: - parent = parents.get(node) - if not (isinstance(parent, ast.Attribute) and parent.attr == "get" - and isinstance(parents.get(parent), ast.Call)): - return expr - node.id = "data" - elif isinstance(node, (ast.Lambda, ast.NamedExpr)): - return expr - return returned - - # -------------------------------------------------------------------------- # inlinability: can `data` escape this body? # -------------------------------------------------------------------------- -def _inlinable(fn: ast.FunctionDef, allow_final_return: bool, any_return: bool = False) -> bool: - body = _body_without_docstring(fn) - if not body: - return not allow_final_return - for node in ast.walk(fn): - if isinstance(node, (ast.Yield, ast.YieldFrom, ast.Await, ast.Global, ast.Nonlocal)): - return False - if isinstance(node, (ast.Lambda, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: - if any(isinstance(n, ast.Name) and n.id == "data" for n in ast.walk(node)): - return False - if isinstance(node, ast.Name) and node.id == "data" and not isinstance(node.ctx, ast.Load): - return False - returns = [n for n in ast.walk(fn) if isinstance(n, ast.Return)] - final = body[-1] if isinstance(body[-1], ast.Return) else None - if any_return: - pass - elif allow_final_return: - if final is None or final.value is None or returns != [final]: - return False - elif returns: - return False - return True - - # -------------------------------------------------------------------------- # names # -------------------------------------------------------------------------- -def _definition_renames(step, prefix: str) -> Dict[str, str]: - renames: Dict[str, str] = {} - for source in step.definitions: - for stmt in ast.parse(source).body: - for name in _top_level_bound(stmt): - renames[name] = prefix + name - return renames - - -def _top_level_bound(stmt: ast.stmt) -> Set[str]: - if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - return {stmt.name} - names: Set[str] = set() - targets = stmt.targets if isinstance(stmt, ast.Assign) else [getattr(stmt, "target", None)] - for target in targets: - for node in ast.walk(target) if target is not None else (): - if isinstance(node, ast.Name): - names.add(node.id) - return names - - -def _rename_names(tree: ast.AST, renames: Dict[str, str]) -> None: - for node in ast.walk(tree): - if isinstance(node, ast.Name) and node.id in renames: - node.id = renames[node.id] - elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name in renames: - node.name = renames[node.name] - - def _renamed_copy(fn: ast.AST, renames: Dict[str, str]) -> ast.AST: fn = copy.deepcopy(fn) - _rename_names(fn, renames) - return fn - - -def _rename_locals(fn: ast.FunctionDef, prefix: str) -> ast.FunctionDef: - """Prefix every name the body binds, so inlined steps cannot collide.""" - fn = copy.deepcopy(fn) - local: Set[str] = set() - for node in ast.walk(fn): - if isinstance(node, ast.Name) and not isinstance(node.ctx, ast.Load): - local.add(node.id) - elif isinstance(node, ast.arg) and node is not fn.args.args[0]: - local.add(node.arg) - elif isinstance(node, ast.ExceptHandler) and node.name: - local.add(node.name) - elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: - local.add(node.name) - local.discard("data") - renames = {name: prefix + name for name in local} - for node in ast.walk(fn): - if isinstance(node, ast.Name) and node.id in renames: - node.id = renames[node.id] - elif isinstance(node, ast.arg) and node.arg in renames: - node.arg = renames[node.arg] - elif isinstance(node, ast.ExceptHandler) and node.name in renames: - node.name = renames[node.name] - elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: - node.name = renames.get(node.name, node.name) + EmitInlineModuleFilter._rename_names(fn, renames) return fn -def _body_without_docstring(fn: ast.FunctionDef) -> List[ast.stmt]: - body = list(fn.body) - if body and isinstance(body[0], ast.Expr) and isinstance(getattr(body[0], "value", None), ast.Constant) \ - and isinstance(body[0].value.value, str): - body = body[1:] - return body - - def _as_function(fn: ast.FunctionDef, name: str) -> str: fn = copy.deepcopy(fn) fn.name = name return ast.unparse(fn) - - -def _merge_import(imports: Dict[str, str], line: str, step) -> None: - bound = _top_level_bound_import(line) - for name in bound: - existing = imports.get(name) - if existing is not None and existing != line: - raise UnsupportedFilterShape( - filter_name=step.name, method_name="imports", line=0, col=0, node_kind="Import", - suggestion=f"`{name}` is imported differently by two steps ({existing!r} vs {line!r}); " - "use the default per-step lowering for this pipeline.", - ) - imports[name] = line - - -def _top_level_bound_import(line: str) -> Set[str]: - stmt = ast.parse(line).body[0] - if isinstance(stmt, ast.Import): - return {alias.asname or alias.name.split(".")[0] for alias in stmt.names} - return {alias.asname or alias.name for alias in stmt.names} diff --git a/packages/codeupipe/codeupipe/compile/filters/emit_orchestrator_filter.py b/packages/codeupipe/codeupipe/compile/filters/emit_orchestrator_filter.py index 655be73..f8a7d54 100644 --- a/packages/codeupipe/codeupipe/compile/filters/emit_orchestrator_filter.py +++ b/packages/codeupipe/codeupipe/compile/filters/emit_orchestrator_filter.py @@ -33,43 +33,6 @@ def _default_config(payload: Payload): } -def _import_lines(steps) -> List[str]: - # Each step lives in its own sibling module, so the orchestrator must - # import it. Relative imports keep step names (``json``, ``truncate``) - # from shadowing top-level modules on the customer's sys.path. - # Code uses each step's generated identifier (``step_03_validate``), never - # its label, reported in State and to Hooks. - lines = [] - for step in steps: - names = [step.ident] - if step.kind == "valve": - names.append(f"{step.ident}_predicate") - lines.append(f"from .{step.ident} import {', '.join(names)}") - return lines - - -def _plain_lines(step, pad: str, var: str = "data") -> List[str]: - """One step, no bookkeeping: the fast form, and a nested pipeline's body - (which works on its own variable, ``var``).""" - fn = step.ident - if step.kind == "valve": - return [f"{pad}if {fn}_predicate({var}):", f"{pad} {var} = {fn}({var})"] - if step.kind == "tap": - # Plane B: a Tap failure never halts the run, matching the dynamic - # runtime. Exception only; KeyboardInterrupt etc. pass. - return [f"{pad}try:", f"{pad} {fn}({var})", f"{pad}except Exception:", f"{pad} pass"] - return [f"{pad}{var} = {fn}({var})"] - - -def _fast_run(steps) -> List[str]: - # Steps write in place, so run() works on its own copy of the caller's dict. - lines = ["def run(data):", " data = dict(data)"] - for step in steps: - lines.extend(_plain_lines(step, " ")) - lines.append(" return data") - return lines - - def _observed_step(name: str, body: List[str], pad: str) -> List[str]: """A Plane A step: hooks around it, timing, the failing-step marker.""" return [ @@ -86,93 +49,6 @@ def _observed_step(name: str, body: List[str], pad: str) -> List[str]: ] -def _observed_run(steps, source_hooks) -> List[str]: - pad = " " - body: List[str] = [] - for group, members in groupby(steps, key=lambda s: s.group): - members = list(members) - if group is not None: - # A nested pipeline: one step in this run's State and Hooks. Its - # own skips and Tap errors belong to its run, as dynamically. - # It works on its own copy, as the nested run does: if it fails, the outer on_error hooks see ``data`` as - # it was when the nested pipeline was entered, as dynamically. - inner = [line for step in members for line in _plain_lines(step, pad, "_nested")] - body.extend(_observed_step(group, [f"{pad}_nested = dict(data)", *inner, f"{pad}data = _nested", - f"{pad}executed.append({group!r})"], pad)) - continue - for step in members: - name, fn = step.name, step.ident - if step.kind == "tap": - body.extend([ - f"{pad}_step = None", - f"{pad}try:", - f"{pad} {fn}(data)", - f"{pad}except Exception as _error:", - f"{pad} errors.append(({name!r}, _error))", - f"{pad}executed.append({name!r})", - ]) - elif step.kind == "valve": - body.extend(_observed_step(name, [ - f"{pad}if {fn}_predicate(data):", - f"{pad} data = {fn}(data)", - f"{pad} executed.append({name!r})", - f"{pad}else:", - f"{pad} skipped.append({name!r})", - ], pad)) - else: - body.extend(_observed_step(name, [ - f"{pad}data = {fn}(data)", - f"{pad}executed.append({name!r})", - ], pad)) - return [ - "from time import perf_counter as _clock", - "", - f"SOURCE_HOOKS = {tuple(source_hooks)!r}", - "", - "", - "def run(data, state=None, hooks=None, timing=False, trace_id=None):", - ' """Run the pipeline, recording State into ``state`` and calling ``hooks``.', - "", - " state: a dict to fill with executed, skipped, errors [(name, error)],", - " timings {name: seconds} (only when ``timing``) and trace_id.", - " hooks: objects with before(step, data), after(step, data) and", - " on_error(step, error, data); step is None for the pipeline itself.", - ' """', - " if hooks is None:", - " if SOURCE_HOOKS:", - " raise TypeError(", - ' "this pipeline was exported with Hooks (" + ", ".join(SOURCE_HOOKS) + "): "', - ' "pass them as run(..., hooks=[...]), or hooks=() to run without them")', - " hooks = ()", - " if state is None:", - " state = {}", - ' executed = state["executed"] = []', - ' skipped = state["skipped"] = []', - ' errors = state["errors"] = []', - ' timings = state["timings"] = {}', - ' state["trace_id"] = trace_id', - " _step = None", - " _t0 = 0.0", - " data = dict(data)", - " for _hook in hooks:", - " _hook.before(None, data)", - " try:", - *(body or [f"{pad}pass"]), - " except Exception as _error:", - " if _step is not None:", - " if timing:", - " timings[_step] = _clock() - _t0", - " for _hook in hooks:", - " _hook.on_error(_step, _error, data)", - " for _hook in hooks:", - " _hook.on_error(None, _error, data)", - " raise", - " for _hook in hooks:", - " _hook.after(None, data)", - " return data", - ] - - class EmitOrchestratorFilter: """Emit the top-level orchestrator that calls each step in order.""" @@ -184,11 +60,11 @@ def call(self, payload: Payload) -> Payload: step_dir = output.get("step", "steps/") files = list(payload.get("files") or []) - lines = _import_lines(pipeline_ir.steps) + lines = EmitOrchestratorFilter._import_lines(pipeline_ir.steps) if payload.get("observe"): - lines.extend(["", ""] + _observed_run(pipeline_ir.steps, payload.get("source_hooks") or ())) + lines.extend(["", ""] + EmitOrchestratorFilter._observed_run(pipeline_ir.steps, payload.get("source_hooks") or ())) else: - lines.extend(["", ""] + _fast_run(pipeline_ir.steps)) + lines.extend(["", ""] + EmitOrchestratorFilter._fast_run(pipeline_ir.steps)) lines.append("") files.append({ @@ -200,3 +76,127 @@ def call(self, payload: Payload) -> Payload: "content": "from .pipeline import run\n\n__all__ = [\"run\"]\n", }) return payload.insert("files", files) + + @staticmethod + def _import_lines(steps) -> List[str]: + # Each step lives in its own sibling module, so the orchestrator must + # import it. Relative imports keep step names (``json``, ``truncate``) + # from shadowing top-level modules on the customer's sys.path. + # Code uses each step's generated identifier (``step_03_validate``), never + # its label, reported in State and to Hooks. + lines = [] + for step in steps: + names = [step.ident] + if step.kind == "valve": + names.append(f"{step.ident}_predicate") + lines.append(f"from .{step.ident} import {', '.join(names)}") + return lines + + @staticmethod + def _plain_lines(step, pad: str, var: str = "data") -> List[str]: + """One step, no bookkeeping: the fast form, and a nested pipeline's body + (which works on its own variable, ``var``).""" + fn = step.ident + if step.kind == "valve": + return [f"{pad}if {fn}_predicate({var}):", f"{pad} {var} = {fn}({var})"] + if step.kind == "tap": + # Plane B: a Tap failure never halts the run, matching the dynamic + # runtime. Exception only; KeyboardInterrupt etc. pass. + return [f"{pad}try:", f"{pad} {fn}({var})", f"{pad}except Exception:", f"{pad} pass"] + return [f"{pad}{var} = {fn}({var})"] + + @staticmethod + def _fast_run(steps) -> List[str]: + # Steps write in place, so run() works on its own copy of the caller's dict. + lines = ["def run(data):", " data = dict(data)"] + for step in steps: + lines.extend(EmitOrchestratorFilter._plain_lines(step, " ")) + lines.append(" return data") + return lines + + @staticmethod + def _observed_run(steps, source_hooks) -> List[str]: + pad = " " + body: List[str] = [] + for group, members in groupby(steps, key=lambda s: s.group): + members = list(members) + if group is not None: + # A nested pipeline: one step in this run's State and Hooks. Its + # own skips and Tap errors belong to its run, as dynamically. + # It works on its own copy, as the nested run does: if it fails, the outer on_error hooks see ``data`` as + # it was when the nested pipeline was entered, as dynamically. + inner = [line for step in members for line in EmitOrchestratorFilter._plain_lines(step, pad, "_nested")] + body.extend(_observed_step(group, [f"{pad}_nested = dict(data)", *inner, f"{pad}data = _nested", + f"{pad}executed.append({group!r})"], pad)) + continue + for step in members: + name, fn = step.name, step.ident + if step.kind == "tap": + body.extend([ + f"{pad}_step = None", + f"{pad}try:", + f"{pad} {fn}(data)", + f"{pad}except Exception as _error:", + f"{pad} errors.append(({name!r}, _error))", + f"{pad}executed.append({name!r})", + ]) + elif step.kind == "valve": + body.extend(_observed_step(name, [ + f"{pad}if {fn}_predicate(data):", + f"{pad} data = {fn}(data)", + f"{pad} executed.append({name!r})", + f"{pad}else:", + f"{pad} skipped.append({name!r})", + ], pad)) + else: + body.extend(_observed_step(name, [ + f"{pad}data = {fn}(data)", + f"{pad}executed.append({name!r})", + ], pad)) + return [ + "from time import perf_counter as _clock", + "", + f"SOURCE_HOOKS = {tuple(source_hooks)!r}", + "", + "", + "def run(data, state=None, hooks=None, timing=False, trace_id=None):", + ' """Run the pipeline, recording State into ``state`` and calling ``hooks``.', + "", + " state: a dict to fill with executed, skipped, errors [(name, error)],", + " timings {name: seconds} (only when ``timing``) and trace_id.", + " hooks: objects with before(step, data), after(step, data) and", + " on_error(step, error, data); step is None for the pipeline itself.", + ' """', + " if hooks is None:", + " if SOURCE_HOOKS:", + " raise TypeError(", + ' "this pipeline was exported with Hooks (" + ", ".join(SOURCE_HOOKS) + "): "', + ' "pass them as run(..., hooks=[...]), or hooks=() to run without them")', + " hooks = ()", + " if state is None:", + " state = {}", + ' executed = state["executed"] = []', + ' skipped = state["skipped"] = []', + ' errors = state["errors"] = []', + ' timings = state["timings"] = {}', + ' state["trace_id"] = trace_id', + " _step = None", + " _t0 = 0.0", + " data = dict(data)", + " for _hook in hooks:", + " _hook.before(None, data)", + " try:", + *(body or [f"{pad}pass"]), + " except Exception as _error:", + " if _step is not None:", + " if timing:", + " timings[_step] = _clock() - _t0", + " for _hook in hooks:", + " _hook.on_error(_step, _error, data)", + " for _hook in hooks:", + " _hook.on_error(None, _error, data)", + " raise", + " for _hook in hooks:", + " _hook.after(None, data)", + " return data", + ] diff --git a/packages/codeupipe/codeupipe/compile/filters/emit_plain_files_filter.py b/packages/codeupipe/codeupipe/compile/filters/emit_plain_files_filter.py index d5a067d..5a5ff38 100644 --- a/packages/codeupipe/codeupipe/compile/filters/emit_plain_files_filter.py +++ b/packages/codeupipe/codeupipe/compile/filters/emit_plain_files_filter.py @@ -46,10 +46,10 @@ def call(self, payload: Payload) -> Payload: # predicate of exactly that name), and the Valve predicate # would then call itself. Move the helper aside instead. wanted = f"{step.ident}_predicate" - if any(wanted in _bound_names(tree) for tree in definitions): - moved = _unused_name(f"{wanted}_impl", definitions) + if any(wanted in EmitPlainFilesFilter._bound_names(tree) for tree in definitions): + moved = EmitPlainFilesFilter._unused_name(f"{wanted}_impl", definitions) for tree in [*definitions, call_ast, predicate]: - _rename(tree, wanted, moved) + EmitPlainFilesFilter._rename(tree, wanted, moved) if isinstance(predicate, ast.FunctionDef): predicate.name = wanted @@ -69,28 +69,28 @@ def call(self, payload: Payload) -> Payload: return payload.insert("files", files) - -def _bound_names(tree: ast.AST) -> set: - names = set() - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - names.add(node.name) - elif isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): - names.add(node.id) - return names - - -def _unused_name(candidate: str, trees) -> str: - taken = set().union(*(_bound_names(t) for t in trees)) if trees else set() - name, n = candidate, 2 - while name in taken: - name, n = f"{candidate}{n}", n + 1 - return name - - -def _rename(tree: ast.AST, old: str, new: str) -> None: - for node in ast.walk(tree): - if isinstance(node, ast.Name) and node.id == old: - node.id = new - elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name == old: - node.name = new + @staticmethod + def _bound_names(tree: ast.AST) -> set: + names = set() + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + names.add(node.name) + elif isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): + names.add(node.id) + return names + + @staticmethod + def _unused_name(candidate: str, trees) -> str: + taken = set().union(*(EmitPlainFilesFilter._bound_names(t) for t in trees)) if trees else set() + name, n = candidate, 2 + while name in taken: + name, n = f"{candidate}{n}", n + 1 + return name + + @staticmethod + def _rename(tree: ast.AST, old: str, new: str) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.Name) and node.id == old: + node.id = new + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name == old: + node.name = new diff --git a/packages/codeupipe/codeupipe/compile/filters/load_pipeline_filter.py b/packages/codeupipe/codeupipe/compile/filters/load_pipeline_filter.py index b6e373b..2aebdf9 100644 --- a/packages/codeupipe/codeupipe/compile/filters/load_pipeline_filter.py +++ b/packages/codeupipe/codeupipe/compile/filters/load_pipeline_filter.py @@ -29,22 +29,6 @@ def _refuse(name: str, node_kind: str, suggestion: str) -> UnsupportedFilterShap node_kind=node_kind, suggestion=suggestion) -def _check_pipeline(pipeline, name: str, allow_hooks: bool = False) -> None: - """Refuse a pipeline whose behaviour static code cannot reproduce. - - ``allow_hooks``: observable export (``observe=True``) calls the top-level - pipeline's Hooks. Hooks on a nested pipeline stay refused: its hooks are - objects owned by that inner run, which the flattened code does not have. - """ - if not hasattr(pipeline, "_steps"): - raise _refuse(name, type(pipeline).__name__, - "Export takes a Pipeline. This object wraps one or is not one.") - if getattr(pipeline, "_hooks", None) and not allow_hooks: - raise _refuse(name, "Hook", - "Hooks do not run in the fast static export. Export with observe=True " - "(top-level Hooks only), or remove them.") - - def _ident(index: int, label: str) -> str: """A safe, readable Python identifier for a step: ``step_03_validate``. @@ -62,35 +46,6 @@ def _is_async(fn) -> bool: return inspect.iscoroutinefunction(fn) or inspect.isasyncgenfunction(fn) -def _flatten(steps, prefix="", group=None): - """Expand nested pipelines into their steps, in order. - - A nested pipeline runs its steps in sequence on the parent's payload, so - lowering it as those steps inline is the same computation. Names get the - nesting step's name as a prefix (``inner__greet``) so they stay unique - and remain unique. Each step also carries ``group``, the - top-level step it came from, so observable export can still report the - nested pipeline as one step. - """ - for name, instance, step_type in steps: - if step_type == "pipeline": - _check_pipeline(instance, f"{prefix}{name}") - yield from _flatten(instance._steps, f"{prefix}{name}/", group or name) - else: - yield f"{prefix}{name}", instance, step_type, group - - -def _refuse_empty_nested(steps) -> None: - """A nested pipeline that flattens to no steps leaves nothing for the - observable export to hang its State entry and Hook calls on, while the - dynamic run still reports it as executed. Refuse rather than drop it.""" - for name, instance, step_type in steps: - if step_type == "pipeline" and not any(True for _ in _flatten(instance._steps)): - raise _refuse(name, "EmptyPipeline", - "A nested pipeline with no steps cannot be reported in State and Hooks " - "by observable export. Remove it, or export without observe=True.") - - class LoadPipelineFilter: """Introspect ``Pipeline._steps`` / ``._hooks`` into a skeleton IR.""" @@ -102,14 +57,14 @@ def call(self, payload: Payload) -> Payload: raise ValueError("observe=True needs the per-step layout: the inline layout writes the " "payload in place, so Hooks would see later steps' writes. Use " "lowering='steps' (the default) to observe.") - _check_pipeline(pipeline, "", allow_hooks=observe) + LoadPipelineFilter._check_pipeline(pipeline, "", allow_hooks=observe) if observe: - _refuse_empty_nested(pipeline._steps) + LoadPipelineFilter._refuse_empty_nested(pipeline._steps) steps = [] step_instances = [] step_predicates = [] - for name, instance, step_type, group in _flatten(pipeline._steps): + for name, instance, step_type, group in LoadPipelineFilter._flatten(pipeline._steps): kind = None source_instance = instance predicate = None @@ -158,3 +113,48 @@ def call(self, payload: Payload) -> Payload: .insert("source_hooks", tuple(type(h).__name__ for h in getattr(pipeline, "_hooks", ()))) ) return result + + @staticmethod + def _check_pipeline(pipeline, name: str, allow_hooks: bool = False) -> None: + """Refuse a pipeline whose behaviour static code cannot reproduce. + + ``allow_hooks``: observable export (``observe=True``) calls the top-level + pipeline's Hooks. Hooks on a nested pipeline stay refused: its hooks are + objects owned by that inner run, which the flattened code does not have. + """ + if not hasattr(pipeline, "_steps"): + raise _refuse(name, type(pipeline).__name__, + "Export takes a Pipeline. This object wraps one or is not one.") + if getattr(pipeline, "_hooks", None) and not allow_hooks: + raise _refuse(name, "Hook", + "Hooks do not run in the fast static export. Export with observe=True " + "(top-level Hooks only), or remove them.") + + @staticmethod + def _flatten(steps, prefix="", group=None): + """Expand nested pipelines into their steps, in order. + + A nested pipeline runs its steps in sequence on the parent's payload, so + lowering it as those steps inline is the same computation. Names get the + nesting step's name as a prefix (``inner__greet``) so they stay unique + and remain unique. Each step also carries ``group``, the + top-level step it came from, so observable export can still report the + nested pipeline as one step. + """ + for name, instance, step_type in steps: + if step_type == "pipeline": + LoadPipelineFilter._check_pipeline(instance, f"{prefix}{name}") + yield from LoadPipelineFilter._flatten(instance._steps, f"{prefix}{name}/", group or name) + else: + yield f"{prefix}{name}", instance, step_type, group + + @staticmethod + def _refuse_empty_nested(steps) -> None: + """A nested pipeline that flattens to no steps leaves nothing for the + observable export to hang its State entry and Hook calls on, while the + dynamic run still reports it as executed. Refuse rather than drop it.""" + for name, instance, step_type in steps: + if step_type == "pipeline" and not any(True for _ in LoadPipelineFilter._flatten(instance._steps)): + raise _refuse(name, "EmptyPipeline", + "A nested pipeline with no steps cannot be reported in State and Hooks " + "by observable export. Remove it, or export without observe=True.") diff --git a/packages/codeupipe/codeupipe/compile/filters/parse_filter_source_filter.py b/packages/codeupipe/codeupipe/compile/filters/parse_filter_source_filter.py index 03d817f..eb0e1e2 100644 --- a/packages/codeupipe/codeupipe/compile/filters/parse_filter_source_filter.py +++ b/packages/codeupipe/codeupipe/compile/filters/parse_filter_source_filter.py @@ -60,13 +60,13 @@ def call(self, payload: Payload) -> Payload: suggestion="Define the step in a real .py file so inspect.getsource can recover it.", ) function_ast = _erase_annotations(ast.parse(source).body[0]) - function_ast, config = _bind_instance_config(function_ast, instance, step) - imports, definitions = _detect_dependencies(function_ast, method, step.name) + function_ast, config = ParseFilterSourceFilter._bind_instance_config(function_ast, instance, step) + imports, definitions = ParseFilterSourceFilter._detect_dependencies(function_ast, method, step.name) definitions = config + definitions predicate_ast = None if step.kind == "valve": - predicate_ast, p_config = _parse_predicate(step, predicate) - p_imports, p_definitions = _detect_dependencies(predicate_ast, predicate.test, step.name) + predicate_ast, p_config = ParseFilterSourceFilter._parse_predicate(step, predicate) + p_imports, p_definitions = ParseFilterSourceFilter._detect_dependencies(predicate_ast, predicate.test, step.name) imports = tuple(sorted(set(imports) | set(p_imports))) p_definitions = p_config + p_definitions definitions = definitions + tuple(d for d in p_definitions if d not in definitions) @@ -83,7 +83,314 @@ def call(self, payload: Payload) -> Payload: return payload.insert("pipeline_ir", replace(pipeline_ir, steps=rewritten_steps)) + @staticmethod + def _bind_instance_config(fn: ast.FunctionDef, instance, step, prefix: str = "_self_") -> Tuple[ast.FunctionDef, tuple]: + """Compile a step's constructor config into it as named constants. + + Export lowers a *live* Pipeline, so each ``self.`` the method reads + already has its value. A value with a literal form is baked in as + ``_self_ = `` — per step, so two instances of one class + keep their own config. Anything that cannot be compiled in is refused + here rather than left as a reference to a ``self`` that will not exist. + """ + if not fn.args.args: + return fn, () + self_name = fn.args.args[0].arg + constants: dict = {} + + def refuse(attr: str, why: str, node: ast.AST) -> UnsupportedFilterShape: + return UnsupportedFilterShape( + filter_name=step.name, + method_name=fn.name, + line=getattr(node, "lineno", 0), + col=getattr(node, "col_offset", 0), + node_kind="Attribute", + suggestion=f"`{self_name}.{attr}` {why}", + ) + + class _Bind(ast.NodeTransformer): + def visit_Attribute(self, node: ast.Attribute): + if not (isinstance(node.value, ast.Name) and node.value.id == self_name): + return self.generic_visit(node) + attr = node.attr + if not isinstance(node.ctx, ast.Load): + raise refuse(attr, "is written inside the step; per-instance state has no static form.", node) + if attr not in constants: + if not hasattr(instance, attr): + raise refuse(attr, "is not set on the instance being exported.", node) + value = getattr(instance, attr) + if callable(value): + raise refuse(attr, "is a method or callable; inline it, or make it a module-level function.", node) + literal = ParseFilterSourceFilter._literal_source(value) + if literal is None: + raise refuse(attr, f"holds a {type(value).__name__}, which has no literal form to compile in.", node) + constants[attr] = (f"{prefix}{attr}", literal) + return ast.copy_location(ast.Name(id=constants[attr][0], ctx=ast.Load()), node) + + def visit_Name(self, node: ast.Name): + if node.id == self_name: + raise refuse("", f"`{self_name}` is used directly; only reads of literal attributes can be compiled in.", node) + return node + + body = [_Bind().visit(stmt) for stmt in fn.body] + fn.body = [ast.fix_missing_locations(stmt) for stmt in body] + definitions = tuple(f"{name} = {literal}" for name, literal in constants.values()) + return fn, definitions + + @staticmethod + def _literal_source(value) -> Optional[str]: + """``repr(value)`` if it evaluates back to an equal value of the same type.""" + source = repr(value) + try: + back = ast.literal_eval(source) + except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError): + return None + if type(back) is not type(value) or back != value: + return None + return source + + @staticmethod + def _parse_predicate(step, predicate) -> Tuple[ast.FunctionDef, tuple]: + """Recover a Valve predicate object's ``test`` as a ``def`` taking ``payload``. + + Predicates are objects (conformance/API.md, "Valve"): the ``test`` + method is lowered like a filter's ``call``, the instance's literal + settings compiled in as constants (``_when_``, so they cannot + collide with the filter's ``_self_`` in the same module). The + orchestrator always emits ``if _predicate(data):``, so a predicate + that cannot be recovered stops the export here. + """ + def refuse(suggestion: str, kind: str = "FunctionDef") -> UnsupportedFilterShape: + return UnsupportedFilterShape( + filter_name=step.name, method_name="predicate", line=0, col=0, node_kind=kind, suggestion=suggestion, + ) + + method = getattr(predicate, "test", None) + if method is None: + raise refuse("A Valve predicate is an object with test(self, payload) -> bool.") + try: + fn = _erase_annotations(ast.parse(textwrap.dedent(inspect.getsource(method))).body[0]) + except (OSError, TypeError, SyntaxError, IndexError): + raise refuse("Define the predicate class in a real .py file so inspect.getsource can recover it.") + if not isinstance(fn, ast.FunctionDef) or len(fn.args.args) != 2: + raise refuse("A Valve predicate's test must be a plain `def test(self, payload)`.") + fn, config = ParseFilterSourceFilter._bind_instance_config(fn, predicate, step, prefix="_when_") + fn.args.args = fn.args.args[1:] # self is gone: its reads are now constants + return ParseFilterSourceFilter._rename_parameter(fn, fn.args.args[0].arg, "payload"), config + + @staticmethod + def _rename_parameter(fn: ast.FunctionDef, old: str, new: str) -> ast.FunctionDef: + """Rename the predicate's parameter so RewritePayloadOps recognises it.""" + if old == new: + return fn + fn.args.args[0].arg = new + for node in ast.walk(fn): + if isinstance(node, ast.Name) and node.id == old: + node.id = new + return fn + + @staticmethod + def _detect_dependencies(fn: ast.FunctionDef, obj, step_name: str) -> Tuple[tuple, tuple]: + """Return ``(imports, definitions)`` the step body needs to run standalone. + + Names are resolved against the defining module's *source* first, because + that is what the author wrote: a top-level import is re-emitted as that + import, and a top-level constant, helper, or class is copied verbatim + (with whatever it needs in turn). Only names the source does not bind + fall back to runtime introspection of the value. + + Resolving by value alone is what dropped same-module constants and + turned ``_WORDS = re.compile(...)`` into ``from re import _WORDS``: a + compiled pattern reports ``__module__ == "re"``. A name that neither + route can place raises instead of being dropped. + """ + module = inspect.getmodule(obj) + if module is None: + return (), () + module_ns = getattr(module, "__dict__", {}) + if not module_ns: + return (), () + bindings = ParseFilterSourceFilter._module_bindings(module) + + imports: set = set() + definitions: dict = {} + pending = list(ParseFilterSourceFilter._referenced_free_names(fn, ParseFilterSourceFilter._local_names(fn))) + seen: set = set() + while pending: + name = pending.pop() + if name in seen or name in _BUILTIN_NAMES: + continue + seen.add(name) + + stmt = bindings.get(name) + if isinstance(stmt, (ast.Import, ast.ImportFrom)): + import_line = ParseFilterSourceFilter._import_line_for_alias(stmt, name) + if import_line is not None: + imports.add(import_line) + continue + elif stmt is not None: + definitions[stmt.lineno] = ast.unparse(stmt) + pending.extend(ParseFilterSourceFilter._free_names(stmt)) + continue + + if name not in module_ns: + continue + import_line = ParseFilterSourceFilter._import_statement_for(name, module_ns[name], module.__name__) + if import_line is None: + raise UnsupportedFilterShape( + filter_name=step_name, + method_name=fn.name, + line=getattr(fn, "lineno", 0), + col=getattr(fn, "col_offset", 0), + node_kind="Name", + suggestion=( + f"`{name}` has no top-level definition or import in {module.__name__}'s source " + "and cannot be imported by name; define it at module level so export can copy it." + ), + ) + imports.add(import_line) + + return tuple(sorted(imports)), tuple(definitions[line] for line in sorted(definitions)) + + @staticmethod + def _module_bindings(module) -> dict: + """Map each name to the last top-level statement in the module that binds it.""" + try: + tree = ast.parse(inspect.getsource(module)) + except (OSError, TypeError, SyntaxError): + return {} + bindings: dict = {} + for stmt in tree.body: + if isinstance(stmt, ast.Import): + for alias in stmt.names: + bindings[alias.asname or alias.name.split(".")[0]] = stmt + elif isinstance(stmt, ast.ImportFrom): + for alias in stmt.names: + bindings[alias.asname or alias.name] = stmt + elif isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + bindings[stmt.name] = stmt + elif isinstance(stmt, ast.Assign): + for target in stmt.targets: + names: set = set() + ParseFilterSourceFilter._collect_target_names(target, names) + for name in names: + bindings[name] = stmt + elif isinstance(stmt, ast.AnnAssign) and stmt.value is not None: + if isinstance(stmt.target, ast.Name): + bindings[stmt.target.id] = stmt + return bindings + + @staticmethod + def _import_line_for_alias(stmt, name: str) -> Optional[str]: + """Re-emit only the alias that binds ``name``; relative imports return None.""" + if isinstance(stmt, ast.ImportFrom): + if stmt.level or stmt.module == "__future__": + return None + for alias in stmt.names: + if (alias.asname or alias.name) == name: + suffix = f" as {alias.asname}" if alias.asname else "" + return f"from {stmt.module} import {alias.name}{suffix}" + return None + for alias in stmt.names: + if (alias.asname or alias.name.split(".")[0]) == name: + return f"import {alias.name} as {alias.asname}" if alias.asname else f"import {alias.name}" + return None + + @staticmethod + def _free_names(stmt: ast.AST) -> set: + """Names a copied top-level statement loads but does not itself bind.""" + loaded: set = set() + bound: set = set() + for node in ast.walk(stmt): + if isinstance(node, ast.Name): + (loaded if isinstance(node.ctx, ast.Load) else bound).add(node.id) + elif isinstance(node, ast.arg): + bound.add(node.arg) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + bound.add(node.name) + return loaded - bound + + @staticmethod + def _local_names(fn: ast.FunctionDef) -> set: + names = {"self", "cls"} + args = fn.args + for arg in list(args.args) + list(args.posonlyargs) + list(args.kwonlyargs): + names.add(arg.arg) + if args.vararg is not None: + names.add(args.vararg.arg) + if args.kwarg is not None: + names.add(args.kwarg.arg) + for node in ast.walk(fn): + if isinstance(node, ast.Assign): + for target in node.targets: + ParseFilterSourceFilter._collect_target_names(target, names) + elif isinstance(node, ast.AugAssign) or isinstance(node, ast.AnnAssign): + ParseFilterSourceFilter._collect_target_names(node.target, names) + elif isinstance(node, (ast.For, ast.comprehension)): + ParseFilterSourceFilter._collect_target_names(node.target, names) + elif isinstance(node, ast.With): + for item in node.items: + if item.optional_vars is not None: + ParseFilterSourceFilter._collect_target_names(item.optional_vars, names) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: + names.add(node.name) + return names + + @staticmethod + def _collect_target_names(target: ast.AST, names: set) -> None: + if isinstance(target, ast.Name): + names.add(target.id) + elif isinstance(target, (ast.Tuple, ast.List)): + for elt in target.elts: + ParseFilterSourceFilter._collect_target_names(elt, names) + elif isinstance(target, ast.Starred): + ParseFilterSourceFilter._collect_target_names(target.value, names) + + @staticmethod + def _referenced_free_names(fn: ast.FunctionDef, local_names: set) -> Iterable[str]: + seen: set = set() + for node in ast.walk(fn): + root = None + if isinstance(node, ast.Name): + root = node.id + elif isinstance(node, ast.Attribute): + root = ParseFilterSourceFilter._leftmost_name(node) + if root and root not in local_names and root not in seen: + seen.add(root) + yield root + + @staticmethod + def _leftmost_name(node: ast.Attribute) -> Optional[str]: + current = node.value + while isinstance(current, ast.Attribute): + current = current.value + if isinstance(current, ast.Name): + return current.id + return None + @staticmethod + def _import_statement_for(local_name: str, value, current_module: str) -> Optional[str]: + if isinstance(value, types.ModuleType): + real_name = value.__name__ + if real_name == local_name: + return f"import {real_name}" + return f"import {real_name} as {local_name}" + + origin = getattr(value, "__module__", None) + if not origin or origin == current_module or origin == "builtins": + return None + real_name = getattr(value, "__name__", local_name) + # ``__module__`` names where a value's *type* lives, not where the value + # can be imported from (a compiled pattern reports "re"). Only emit the + # import if it would actually hand back this object. + if getattr(sys.modules.get(origin), real_name, None) is not value: + return None + if real_name == local_name: + return f"from {origin} import {local_name}" + return f"from {origin} import {real_name} as {local_name}" + + +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class _EraseAnnotations(ast.NodeTransformer): """Drop every type hint in a function: on its parameters, its return, and its locals (``x: T = v`` becomes ``x = v``; a bare ``x: T`` goes).""" @@ -119,309 +426,3 @@ def _erase_annotations(fn): never refers to the hint's names (``Payload``, a TypedDict) and so never imports the framework.""" return ast.fix_missing_locations(_EraseAnnotations().visit(fn)) - - -def _bind_instance_config(fn: ast.FunctionDef, instance, step, prefix: str = "_self_") -> Tuple[ast.FunctionDef, tuple]: - """Compile a step's constructor config into it as named constants. - - Export lowers a *live* Pipeline, so each ``self.`` the method reads - already has its value. A value with a literal form is baked in as - ``_self_ = `` — per step, so two instances of one class - keep their own config. Anything that cannot be compiled in is refused - here rather than left as a reference to a ``self`` that will not exist. - """ - if not fn.args.args: - return fn, () - self_name = fn.args.args[0].arg - constants: dict = {} - - def refuse(attr: str, why: str, node: ast.AST) -> UnsupportedFilterShape: - return UnsupportedFilterShape( - filter_name=step.name, - method_name=fn.name, - line=getattr(node, "lineno", 0), - col=getattr(node, "col_offset", 0), - node_kind="Attribute", - suggestion=f"`{self_name}.{attr}` {why}", - ) - - class _Bind(ast.NodeTransformer): - def visit_Attribute(self, node: ast.Attribute): - if not (isinstance(node.value, ast.Name) and node.value.id == self_name): - return self.generic_visit(node) - attr = node.attr - if not isinstance(node.ctx, ast.Load): - raise refuse(attr, "is written inside the step; per-instance state has no static form.", node) - if attr not in constants: - if not hasattr(instance, attr): - raise refuse(attr, "is not set on the instance being exported.", node) - value = getattr(instance, attr) - if callable(value): - raise refuse(attr, "is a method or callable; inline it, or make it a module-level function.", node) - literal = _literal_source(value) - if literal is None: - raise refuse(attr, f"holds a {type(value).__name__}, which has no literal form to compile in.", node) - constants[attr] = (f"{prefix}{attr}", literal) - return ast.copy_location(ast.Name(id=constants[attr][0], ctx=ast.Load()), node) - - def visit_Name(self, node: ast.Name): - if node.id == self_name: - raise refuse("", f"`{self_name}` is used directly; only reads of literal attributes can be compiled in.", node) - return node - - body = [_Bind().visit(stmt) for stmt in fn.body] - fn.body = [ast.fix_missing_locations(stmt) for stmt in body] - definitions = tuple(f"{name} = {literal}" for name, literal in constants.values()) - return fn, definitions - - -def _literal_source(value) -> Optional[str]: - """``repr(value)`` if it evaluates back to an equal value of the same type.""" - source = repr(value) - try: - back = ast.literal_eval(source) - except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError): - return None - if type(back) is not type(value) or back != value: - return None - return source - - -def _parse_predicate(step, predicate) -> Tuple[ast.FunctionDef, tuple]: - """Recover a Valve predicate object's ``test`` as a ``def`` taking ``payload``. - - Predicates are objects (conformance/API.md, "Valve"): the ``test`` - method is lowered like a filter's ``call``, the instance's literal - settings compiled in as constants (``_when_``, so they cannot - collide with the filter's ``_self_`` in the same module). The - orchestrator always emits ``if _predicate(data):``, so a predicate - that cannot be recovered stops the export here. - """ - def refuse(suggestion: str, kind: str = "FunctionDef") -> UnsupportedFilterShape: - return UnsupportedFilterShape( - filter_name=step.name, method_name="predicate", line=0, col=0, node_kind=kind, suggestion=suggestion, - ) - - method = getattr(predicate, "test", None) - if method is None: - raise refuse("A Valve predicate is an object with test(self, payload) -> bool.") - try: - fn = _erase_annotations(ast.parse(textwrap.dedent(inspect.getsource(method))).body[0]) - except (OSError, TypeError, SyntaxError, IndexError): - raise refuse("Define the predicate class in a real .py file so inspect.getsource can recover it.") - if not isinstance(fn, ast.FunctionDef) or len(fn.args.args) != 2: - raise refuse("A Valve predicate's test must be a plain `def test(self, payload)`.") - fn, config = _bind_instance_config(fn, predicate, step, prefix="_when_") - fn.args.args = fn.args.args[1:] # self is gone: its reads are now constants - return _rename_parameter(fn, fn.args.args[0].arg, "payload"), config - - -def _rename_parameter(fn: ast.FunctionDef, old: str, new: str) -> ast.FunctionDef: - """Rename the predicate's parameter so RewritePayloadOps recognises it.""" - if old == new: - return fn - fn.args.args[0].arg = new - for node in ast.walk(fn): - if isinstance(node, ast.Name) and node.id == old: - node.id = new - return fn - - -def _detect_dependencies(fn: ast.FunctionDef, obj, step_name: str) -> Tuple[tuple, tuple]: - """Return ``(imports, definitions)`` the step body needs to run standalone. - - Names are resolved against the defining module's *source* first, because - that is what the author wrote: a top-level import is re-emitted as that - import, and a top-level constant, helper, or class is copied verbatim - (with whatever it needs in turn). Only names the source does not bind - fall back to runtime introspection of the value. - - Resolving by value alone is what dropped same-module constants and - turned ``_WORDS = re.compile(...)`` into ``from re import _WORDS``: a - compiled pattern reports ``__module__ == "re"``. A name that neither - route can place raises instead of being dropped. - """ - module = inspect.getmodule(obj) - if module is None: - return (), () - module_ns = getattr(module, "__dict__", {}) - if not module_ns: - return (), () - bindings = _module_bindings(module) - - imports: set = set() - definitions: dict = {} - pending = list(_referenced_free_names(fn, _local_names(fn))) - seen: set = set() - while pending: - name = pending.pop() - if name in seen or name in _BUILTIN_NAMES: - continue - seen.add(name) - - stmt = bindings.get(name) - if isinstance(stmt, (ast.Import, ast.ImportFrom)): - import_line = _import_line_for_alias(stmt, name) - if import_line is not None: - imports.add(import_line) - continue - elif stmt is not None: - definitions[stmt.lineno] = ast.unparse(stmt) - pending.extend(_free_names(stmt)) - continue - - if name not in module_ns: - continue - import_line = _import_statement_for(name, module_ns[name], module.__name__) - if import_line is None: - raise UnsupportedFilterShape( - filter_name=step_name, - method_name=fn.name, - line=getattr(fn, "lineno", 0), - col=getattr(fn, "col_offset", 0), - node_kind="Name", - suggestion=( - f"`{name}` has no top-level definition or import in {module.__name__}'s source " - "and cannot be imported by name; define it at module level so export can copy it." - ), - ) - imports.add(import_line) - - return tuple(sorted(imports)), tuple(definitions[line] for line in sorted(definitions)) - - -def _module_bindings(module) -> dict: - """Map each name to the last top-level statement in the module that binds it.""" - try: - tree = ast.parse(inspect.getsource(module)) - except (OSError, TypeError, SyntaxError): - return {} - bindings: dict = {} - for stmt in tree.body: - if isinstance(stmt, ast.Import): - for alias in stmt.names: - bindings[alias.asname or alias.name.split(".")[0]] = stmt - elif isinstance(stmt, ast.ImportFrom): - for alias in stmt.names: - bindings[alias.asname or alias.name] = stmt - elif isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - bindings[stmt.name] = stmt - elif isinstance(stmt, ast.Assign): - for target in stmt.targets: - names: set = set() - _collect_target_names(target, names) - for name in names: - bindings[name] = stmt - elif isinstance(stmt, ast.AnnAssign) and stmt.value is not None: - if isinstance(stmt.target, ast.Name): - bindings[stmt.target.id] = stmt - return bindings - - -def _import_line_for_alias(stmt, name: str) -> Optional[str]: - """Re-emit only the alias that binds ``name``; relative imports return None.""" - if isinstance(stmt, ast.ImportFrom): - if stmt.level or stmt.module == "__future__": - return None - for alias in stmt.names: - if (alias.asname or alias.name) == name: - suffix = f" as {alias.asname}" if alias.asname else "" - return f"from {stmt.module} import {alias.name}{suffix}" - return None - for alias in stmt.names: - if (alias.asname or alias.name.split(".")[0]) == name: - return f"import {alias.name} as {alias.asname}" if alias.asname else f"import {alias.name}" - return None - - -def _free_names(stmt: ast.AST) -> set: - """Names a copied top-level statement loads but does not itself bind.""" - loaded: set = set() - bound: set = set() - for node in ast.walk(stmt): - if isinstance(node, ast.Name): - (loaded if isinstance(node.ctx, ast.Load) else bound).add(node.id) - elif isinstance(node, ast.arg): - bound.add(node.arg) - elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - bound.add(node.name) - return loaded - bound - - -def _local_names(fn: ast.FunctionDef) -> set: - names = {"self", "cls"} - args = fn.args - for arg in list(args.args) + list(args.posonlyargs) + list(args.kwonlyargs): - names.add(arg.arg) - if args.vararg is not None: - names.add(args.vararg.arg) - if args.kwarg is not None: - names.add(args.kwarg.arg) - for node in ast.walk(fn): - if isinstance(node, ast.Assign): - for target in node.targets: - _collect_target_names(target, names) - elif isinstance(node, ast.AugAssign) or isinstance(node, ast.AnnAssign): - _collect_target_names(node.target, names) - elif isinstance(node, (ast.For, ast.comprehension)): - _collect_target_names(node.target, names) - elif isinstance(node, ast.With): - for item in node.items: - if item.optional_vars is not None: - _collect_target_names(item.optional_vars, names) - elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node is not fn: - names.add(node.name) - return names - - -def _collect_target_names(target: ast.AST, names: set) -> None: - if isinstance(target, ast.Name): - names.add(target.id) - elif isinstance(target, (ast.Tuple, ast.List)): - for elt in target.elts: - _collect_target_names(elt, names) - elif isinstance(target, ast.Starred): - _collect_target_names(target.value, names) - - -def _referenced_free_names(fn: ast.FunctionDef, local_names: set) -> Iterable[str]: - seen: set = set() - for node in ast.walk(fn): - root = None - if isinstance(node, ast.Name): - root = node.id - elif isinstance(node, ast.Attribute): - root = _leftmost_name(node) - if root and root not in local_names and root not in seen: - seen.add(root) - yield root - - -def _leftmost_name(node: ast.Attribute) -> Optional[str]: - current = node.value - while isinstance(current, ast.Attribute): - current = current.value - if isinstance(current, ast.Name): - return current.id - return None - - -def _import_statement_for(local_name: str, value, current_module: str) -> Optional[str]: - if isinstance(value, types.ModuleType): - real_name = value.__name__ - if real_name == local_name: - return f"import {real_name}" - return f"import {real_name} as {local_name}" - - origin = getattr(value, "__module__", None) - if not origin or origin == current_module or origin == "builtins": - return None - real_name = getattr(value, "__name__", local_name) - # ``__module__`` names where a value's *type* lives, not where the value - # can be imported from (a compiled pattern reports "re"). Only emit the - # import if it would actually hand back this object. - if getattr(sys.modules.get(origin), real_name, None) is not value: - return None - if real_name == local_name: - return f"from {origin} import {local_name}" - return f"from {origin} import {real_name} as {local_name}" diff --git a/packages/codeupipe/codeupipe/compile/filters/rewrite_payload_ops_filter.py b/packages/codeupipe/codeupipe/compile/filters/rewrite_payload_ops_filter.py index 181bc27..951e30b 100644 --- a/packages/codeupipe/codeupipe/compile/filters/rewrite_payload_ops_filter.py +++ b/packages/codeupipe/codeupipe/compile/filters/rewrite_payload_ops_filter.py @@ -36,21 +36,6 @@ def _name(identifier: str, ctx=None) -> ast.Name: _FRAMEWORK_MODULES = ("codeupipe", "codeupipe") -def _drop_unused_framework_imports(imports: tuple, *trees: ast.AST) -> tuple: - """Lowering removes ``Payload`` uses; drop their imports - so no framework import survives into the emitted package.""" - used = {n.id for tree in trees if tree is not None for n in ast.walk(tree) if isinstance(n, ast.Name)} - kept = [] - for line in imports: - stmt = ast.parse(line).body[0] - module = getattr(stmt, "module", None) or (stmt.names[0].name if isinstance(stmt, ast.Import) else "") - bound = {a.asname or a.name.split(".")[0] for a in stmt.names} - if module.split(".")[0] in _FRAMEWORK_MODULES and not (bound & used): - continue - kept.append(line) - return tuple(kept) - - # Payload members this rewriter lowers. Anything else read off the payload # (``trace_id``, ``lineage``, ``with_trace``, ``_data``) has no plain-dict # counterpart and would otherwise pass through as ``data.`` on a dict, @@ -67,6 +52,7 @@ def _drop_unused_framework_imports(imports: tuple, *trees: ast.AST) -> tuple: "of a return, or as the value assigned to a name. Move it out of the larger expression.") +# cup: allow-plain a visitor of a syntax tree: the tool that a Filter uses to read or change source code class _PayloadToDictTransformer(ast.NodeTransformer): """Lower a step's Payload operations to operations on a plain dict. @@ -281,7 +267,22 @@ def call(self, payload: Payload) -> Payload: predicate_ast = ast.fix_missing_locations( predicate_transformer.visit(ast.parse(ast.unparse(predicate_ast)).body[0]) ) - imports = _drop_unused_framework_imports(tuple(step.imports or ()), call_ast, predicate_ast) + imports = RewritePayloadOpsFilter._drop_unused_framework_imports(tuple(step.imports or ()), call_ast, predicate_ast) rewritten_steps.append(replace(step, call_ast=call_ast, predicate_ast=predicate_ast, imports=imports)) return payload.insert("pipeline_ir", replace(pipeline_ir, steps=rewritten_steps)) + + @staticmethod + def _drop_unused_framework_imports(imports: tuple, *trees: ast.AST) -> tuple: + """Lowering removes ``Payload`` uses; drop their imports + so no framework import survives into the emitted package.""" + used = {n.id for tree in trees if tree is not None for n in ast.walk(tree) if isinstance(n, ast.Name)} + kept = [] + for line in imports: + stmt = ast.parse(line).body[0] + module = getattr(stmt, "module", None) or (stmt.names[0].name if isinstance(stmt, ast.Import) else "") + bound = {a.asname or a.name.split(".")[0] for a in stmt.names} + if module.split(".")[0] in _FRAMEWORK_MODULES and not (bound & used): + continue + kept.append(line) + return tuple(kept) diff --git a/packages/codeupipe/codeupipe/compile/ir.py b/packages/codeupipe/codeupipe/compile/ir.py index bd09b1e..88d7a64 100644 --- a/packages/codeupipe/codeupipe/compile/ir.py +++ b/packages/codeupipe/codeupipe/compile/ir.py @@ -92,7 +92,7 @@ class SymbolFlowShapeIR: @dataclass(frozen=True) -class StepIR: +class StepIR: # cup: allow-plain a record of data: it checks its own fields when it is made """A single pipeline step in canonical form. Attributes: diff --git a/packages/codeupipe/codeupipe/compile/pipelines/export_pipeline.py b/packages/codeupipe/codeupipe/compile/pipelines/export_pipeline.py index a4ca9f6..96a6984 100644 --- a/packages/codeupipe/codeupipe/compile/pipelines/export_pipeline.py +++ b/packages/codeupipe/codeupipe/compile/pipelines/export_pipeline.py @@ -19,10 +19,10 @@ from .lowering_mode_predicate import LoweringModePredicate -__all__ = ["build_export_pipeline_v2"] +__all__ = ["ExportPipeline", "build_export_pipeline_v2"] -def build_export_pipeline_v2(*, write: bool = True) -> Pipeline: +class ExportPipeline(Pipeline): """Assemble the v2 export pipeline. Args: @@ -48,14 +48,20 @@ def build_export_pipeline_v2(*, write: bool = True) -> Pipeline: ``pipeline_ir`` (PipelineIR) — the canonical IR. ``files`` (list[dict]) — the emitted files. """ - p = Pipeline() - p.add_filter(LoadPipelineFilter()) - p.add_filter(ParseFilterSourceFilter()) - p.add_filter(RewritePayloadOpsFilter()) - p.add_filter(LowerControlFlowFilter()) - p.add_valve(LoweringModePredicate(inline=False), EmitPlainFilesFilter()) - p.add_valve(LoweringModePredicate(inline=False), EmitOrchestratorFilter()) - p.add_valve(LoweringModePredicate(inline=True), EmitInlineModuleFilter()) - if write: - p.add_tap(WriteFilesTap()) - return p + + def __init__(self, *, write: bool = True): + super().__init__() + self.add_filter(LoadPipelineFilter()) + self.add_filter(ParseFilterSourceFilter()) + self.add_filter(RewritePayloadOpsFilter()) + self.add_filter(LowerControlFlowFilter()) + self.add_valve(LoweringModePredicate(inline=False), EmitPlainFilesFilter()) + self.add_valve(LoweringModePredicate(inline=False), EmitOrchestratorFilter()) + self.add_valve(LoweringModePredicate(inline=True), EmitInlineModuleFilter()) + if write: + self.add_tap(WriteFilesTap()) + + +def build_export_pipeline_v2(*, write: bool = True) -> Pipeline: + """The export pipeline. Kept for callers that used this name: see ``ExportPipeline``.""" + return ExportPipeline(write=write) diff --git a/packages/codeupipe/codeupipe/compile/resolve_target.py b/packages/codeupipe/codeupipe/compile/resolve_target.py index d5c3845..fbe450d 100644 --- a/packages/codeupipe/codeupipe/compile/resolve_target.py +++ b/packages/codeupipe/codeupipe/compile/resolve_target.py @@ -14,6 +14,7 @@ class TargetError(Exception): """The ``module:name`` given does not lead to a Pipeline.""" +# cup: allow-plain used by the entry of the program: it finds the pipeline that an argument names def resolve_target(spec: str) -> Pipeline: """``name`` may be a Pipeline, or anything callable with no arguments that returns one: a builder function or a named Pipeline type.""" diff --git a/packages/codeupipe/codeupipe/config.py b/packages/codeupipe/codeupipe/config.py index 35ce5f1..847cd8b 100644 --- a/packages/codeupipe/codeupipe/config.py +++ b/packages/codeupipe/codeupipe/config.py @@ -27,6 +27,7 @@ class ConfigError(Exception): """The config file cannot be used as it is.""" +# cup: allow-plain it reads the config file, before any pipeline exists def _unknown(where: str, given, allowed) -> None: extra = sorted(set(given) - set(allowed)) if extra: @@ -34,11 +35,13 @@ def _unknown(where: str, given, allowed) -> None: where, ", ".join(repr(key) for key in extra), ", ".join(allowed))) +# cup: allow-plain it reads the config file, before any pipeline exists def _strings(where: str, value) -> None: if not (isinstance(value, list) and all(isinstance(item, str) for item in value)): raise ConfigError("%s must be a list of strings" % where) +# cup: allow-plain it reads the config file, before any pipeline exists def load_config(path: Optional[str] = None) -> Optional[dict]: """The config as a dict, checked. With no ``path``: ``cup.config.json`` in the working directory, or None when there is no such file.""" diff --git a/packages/codeupipe/codeupipe/core/payload.py b/packages/codeupipe/codeupipe/core/payload.py index 5a9be90..82c14f5 100644 --- a/packages/codeupipe/codeupipe/core/payload.py +++ b/packages/codeupipe/codeupipe/core/payload.py @@ -13,6 +13,7 @@ T = TypeVar('T') +# cup: allow-plain a primitive itself: the framework makes the primitives, so it is not written with them class Payload(Generic[T]): """Keyed data. Keys are strings; values are JSON-shaped. diff --git a/packages/codeupipe/codeupipe/core/pipeline.py b/packages/codeupipe/codeupipe/core/pipeline.py index b1a1824..c3b4d1e 100644 --- a/packages/codeupipe/codeupipe/core/pipeline.py +++ b/packages/codeupipe/codeupipe/core/pipeline.py @@ -24,6 +24,7 @@ TOutput = TypeVar('TOutput') +# cup: allow-plain a primitive itself: the framework makes the primitives, so it is not written with them class Pipeline(Generic[TInput, TOutput]): """ Orchestrator — runs filters in sequence with hooks, taps, and state tracking. diff --git a/packages/codeupipe/codeupipe/core/step_label.py b/packages/codeupipe/codeupipe/core/step_label.py index bac8e34..9195c10 100644 --- a/packages/codeupipe/codeupipe/core/step_label.py +++ b/packages/codeupipe/codeupipe/core/step_label.py @@ -9,6 +9,7 @@ __all__ = ["label_of"] +# cup: allow-plain a part of the Pipeline primitive: it gives a step its label def label_of(step: Any) -> str: """The class name of ``step``; for a Valve, of its Filter.""" inner = getattr(step, "_inner", None) diff --git a/scorecard/case_gen.py b/scorecard/case_gen.py index 3302630..176e475 100644 --- a/scorecard/case_gen.py +++ b/scorecard/case_gen.py @@ -46,6 +46,7 @@ ] +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def generate_value(rng, depth=0): """A JSON-shaped value (None, bool, int, float, str, list, dict).""" roll = rng.random() @@ -68,6 +69,7 @@ def _add_operand(rng): return rng.choice([1, 1, 2, -3, 0.5, 10 ** 20, True, "s", [1], None]) +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _filter_spec(rng): op = rng.choices(FILTER_OPS, weights=[5, 3, 4, 3, 3, 2, 2, 2, 1])[0] key = rng.choice(KEYS) @@ -90,12 +92,14 @@ def _filter_spec(rng): return {"op": "fail", "message": rng.choice(["boom", "bad 'input'", "ünïcode"])} +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _when(rng): if rng.random() < 0.5: return {"op": "truthy", "key": rng.choice(KEYS)} return {"op": "equals", "key": rng.choice(KEYS), "value": rng.choice([0, 1, True, None, "x", "", [1], {"a": 1}])} +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _namer(rng, hostile): """next_name(prefix): unique ``prefix``, or sometimes a hostile name.""" counter = [0] @@ -108,6 +112,7 @@ def next_name(prefix): return next_name +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _steps(rng, names, depth, max_steps, max_depth): out = [] for _ in range(rng.randint(0, max_steps)): @@ -131,6 +136,7 @@ def _steps(rng, names, depth, max_steps, max_depth): return out +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def generate_case(rng, index=0, max_steps=8, max_depth=2, hostile_rate=0.0): """One random case. ``rng`` is a ``random.Random``; same seed, same case. diff --git a/scorecard/compare.py b/scorecard/compare.py index c69e35c..071d0d1 100644 --- a/scorecard/compare.py +++ b/scorecard/compare.py @@ -10,12 +10,14 @@ __all__ = ["divergences", "self_checks", "verdict"] +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _diff(check, a, b, out): detail = first_difference(a, b) if detail: out.append({"check": check, "detail": detail}) +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def self_checks(record, who): """Checks a run makes of itself: input untouched, reruns identical.""" out = [] @@ -26,6 +28,7 @@ def self_checks(record, who): return out +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def divergences(dynamic, record, mode): out = [] if record["status"] == "crash": @@ -47,6 +50,7 @@ def divergences(dynamic, record, mode): return out +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def verdict(record, found): if found: return "fail" diff --git a/scorecard/differential.py b/scorecard/differential.py index 68a9a2a..d9a6b31 100644 --- a/scorecard/differential.py +++ b/scorecard/differential.py @@ -31,6 +31,7 @@ RESULTS = pathlib.Path(__file__).resolve().parent / "results" +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def check_case(case): """{"dynamic": [self-check failures], "modes": {mode: {verdict, divergences, reason}}}""" predicates = vocab.make_predicates(case["steps"], vocab.new_workdir()) @@ -44,6 +45,7 @@ def check_case(case): return result +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def failure_signature(result): """What a reproducer must keep failing on: (mode, check) pairs.""" sig = {("dynamic", d["check"]) for d in result["dynamic"]} @@ -52,6 +54,7 @@ def failure_signature(result): return sig +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _refusal_kind(reason): """'' from 'name.call at line 0:0 — cannot lower . Hint: ...'""" if reason and "cannot lower " in reason: @@ -59,6 +62,7 @@ def _refusal_kind(reason): return reason or "?" +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def run_campaign(seed, count, hostile_rate=0.15, max_repros=12, max_steps=8, max_depth=2): started = time.time() cases = case_gen.generate_cases(seed, count, max_steps=max_steps, max_depth=max_depth, @@ -120,6 +124,7 @@ def run_campaign(seed, count, hostile_rate=0.15, max_repros=12, max_steps=8, max } +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def markdown(report): lines = [f"# Differential run: seed {report['seed']}, {report['count']} cases", "", f"{report['seconds']} s. Corpus: " + ", ".join(f"{k} {v}" for k, v in report["corpus"].items()) + ".", @@ -152,6 +157,7 @@ def markdown(report): return "\n".join(lines) + "\n" +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) parser.add_argument("--seed", type=int, default=20260927) diff --git a/scorecard/minimize.py b/scorecard/minimize.py index 6c3ac5d..ec76268 100644 --- a/scorecard/minimize.py +++ b/scorecard/minimize.py @@ -15,6 +15,7 @@ __all__ = ["minimize", "candidates"] +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _step_lists(steps, path=()): """(path, list) for the top-level list and every nested pipeline's list.""" yield path, steps @@ -23,6 +24,7 @@ def _step_lists(steps, path=()): yield from _step_lists(step["steps"], path + (i,)) +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _resolve(case, path): steps = case["steps"] for i in path: @@ -30,6 +32,7 @@ def _resolve(case, path): return steps +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def candidates(case): """Smaller variants of ``case``, most aggressive first.""" for path, steps in list(_step_lists(case["steps"])): @@ -69,6 +72,7 @@ def candidates(case): yield variant +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def minimize(case, still_fails, max_attempts=400): current = copy.deepcopy(case) attempts = 0 diff --git a/scorecard/modes.py b/scorecard/modes.py index a556506..5b22468 100644 --- a/scorecard/modes.py +++ b/scorecard/modes.py @@ -35,6 +35,7 @@ def describe_error(error): return f"{type(error).__name__}: {error}" +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _call(fn, data): vocab.sink.reset() try: @@ -55,6 +56,7 @@ def _run_twice(fn, case): return record +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def _export(pipeline, lowering, observe): try: return vocab.load_export(pipeline, lowering, observe, vocab.new_workdir()), None @@ -90,6 +92,7 @@ def fn(data): return record +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def run_fast(case, predicates, lowering): """Fast export. A case with Hooks must be *refused* (documented); the same pipeline without Hooks is then exported and compared on output, @@ -110,6 +113,7 @@ def run_fast(case, predicates, lowering): return record +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def run_observe(case, predicates): """Observable export. The pipeline carries the case's Hooks (so SOURCE_HOOKS is exercised); a SnapshotHook is passed at run time diff --git a/scorecard/snapshot_hook.py b/scorecard/snapshot_hook.py index ac03bfe..a78c64f 100644 --- a/scorecard/snapshot_hook.py +++ b/scorecard/snapshot_hook.py @@ -15,18 +15,13 @@ __all__ = ["SnapshotHook"] -def _as_dict(payload): - to_dict = getattr(payload, "to_dict", None) - return to_dict() if callable(to_dict) else payload - - class SnapshotHook: def __init__(self): self.events = [] # (event, step, snapshot, error or None) def _record(self, event, step, payload, error=None): err = None if error is None else f"{type(error).__name__}: {error}" - self.events.append((event, step, copy.deepcopy(_as_dict(payload)), err)) + self.events.append((event, step, copy.deepcopy(SnapshotHook._as_dict(payload)), err)) def before(self, step, payload): self._record("before", step, payload) @@ -41,3 +36,7 @@ def sequence(self): """[(event, step, snapshot, error)] for parity comparison.""" return list(self.events) + @staticmethod + def _as_dict(payload): + to_dict = getattr(payload, "to_dict", None) + return to_dict() if callable(to_dict) else payload diff --git a/scorecard/strict.py b/scorecard/strict.py index c7276d7..dc19e6f 100644 --- a/scorecard/strict.py +++ b/scorecard/strict.py @@ -8,6 +8,7 @@ __all__ = ["strict_equal", "first_difference"] +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def first_difference(a, b, path="$"): """None if ``a`` and ``b`` are strictly equal, else a description of the first place they differ.""" diff --git a/scorecard/vocab.py b/scorecard/vocab.py index b775eba..a2790b5 100644 --- a/scorecard/vocab.py +++ b/scorecard/vocab.py @@ -19,6 +19,7 @@ ROOT = pathlib.Path(__file__).resolve().parents[1] RUNNER_DIR = ROOT / "conformance" / "runners" / "python" +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) if str(RUNNER_DIR) not in sys.path: @@ -40,6 +41,7 @@ def new_workdir(prefix="cup_scorecard_"): return pathlib.Path(tempfile.mkdtemp(prefix=prefix)) +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def walk_steps(steps): """Every step spec, depth first, nested pipelines included.""" for step in steps: @@ -48,6 +50,7 @@ def walk_steps(steps): yield from walk_steps(step["steps"]) +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def make_predicates(steps, workdir): """A namespace with ``when_`` for every Valve, each a real ``def`` in a real file (static lowering needs its source).""" @@ -66,6 +69,7 @@ def make_predicates(steps, workdir): return namespace +# cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def build_pipeline(case, predicates, hooks=()): """The case as a live codeupipe Pipeline (runner.build), with ``hooks``.""" pipeline = runner.build(case, Pipeline, predicates, {}) diff --git a/tests/lint/invariants/test_repo_is_clean.py b/tests/lint/invariants/test_repo_is_clean.py index 53fc856..5456e21 100644 --- a/tests/lint/invariants/test_repo_is_clean.py +++ b/tests/lint/invariants/test_repo_is_clean.py @@ -6,12 +6,20 @@ from ._support import REPO +# The places with a `cup: allow-plain` comment. +PLAIN_BY_DESIGN = 81 + CHECKED = ["packages/codeupipe/codeupipe", "conformance", "scorecard", "packages/codeupipe-lint/codeupipe_lint", "tests", "examples"] def test_repo_python_has_no_invariant_violations(): out = lint_paths([os.path.join(REPO, p) for p in CHECKED], exclude=["*/tests/lint/fixtures/*"]) assert out.get("errors") == [] - # CUP-I008 is a warning, and the framework has logic that cannot be a primitive: it makes the primitives. - assert [f.format() for f in out.get("findings") if f.rule != "CUP-I008"] == [] - assert out.get("suppressed") == [] + # No finding and no warning: the repository passes its own linter in strict mode. + assert [f.format() for f in out.get("findings")] == [] + # The only comments that allow a finding are for CUP-I008, on code that is plain for a reason that + # the comment gives: a primitive itself, a visitor of a syntax tree, the entry of a program, a test + # harness. The number is here so that a new one is a change that someone sees. + allowed = out.get("suppressed") + assert {f.rule for f in allowed} == {"CUP-I008"} + assert len(allowed) == PLAIN_BY_DESIGN, sorted({f.path for f in allowed})