fix(mypy): fix orchestrator/serializer type gaps, scope enricher pattern overrides

Enrichers declare InputType/OutputType as plain class attributes
(NotImplemented on the base, overridden per subclass) and use them in
annotations like List[InputType] and self.data.domain — the documented,
intentional API (see flowsint-enricher-builder skill), not something
to individually silence. mypy structurally can't verify a class
attribute doubling as a type alias without making Enricher generic,
which would mean touching every enricher's class declaration. Narrows
the specific checks that pattern trips (valid-type, attr-defined,
name-defined) via per-module overrides in flowsint-enrichers and
flowsint-core's tests (MockEnricher fixtures use the same pattern).

Separately, fixes real gaps in orchestrator.py surfaced once the noise
cleared: several dict literals (step_result, log_entry, results,
results_mapping, branch_results) were inferred as narrower types than
what the code actually assigns into them; execution_log_file and the
enrichers map were missing their real (Optional) types; _create_execution_log
used self.execution_log_file across an assignment mypy couldn't narrow,
fixed via a local variable. update_results_mapping's outputs param is
typed to match what it can actually receive (dict or list, per the
isinstance check upstream) with a comment on the pre-existing dict-only
assumption in its body — not silently "fixed" since that would change
behavior for list-shaped enricher outputs, unverified here. scan()'s
Liskov-incompatible override (sync, single-dict vs the base's async/
list-of-dicts contract) is documented and suppressed, not changed —
that's a behavioral call about how FlowOrchestrator is actually
invoked. graph_edge_to_neo4j_dict/serializer.py's from_obj/to_obj were
typed as generic BaseModel while the body only ever handles
FlowsintType or GraphNode — narrowed to that Union. And
prepare_enricher_inputs, which has no callers anywhere, is typed Any
rather than guessing at intended behavior for dead code whose two
return branches already disagree.

Verified: flowsint-core's 397 tests pass unchanged.
This commit is contained in:
dextmorgn
2026-08-25 09:33:36 +02:00
parent 3a0ca88e10
commit fa1c8e7580
5 changed files with 73 additions and 21 deletions
+3 -1
View File
@@ -21,7 +21,9 @@ if not database_url:
raise RuntimeError("DATABASE_URL is not defined in .env")
config.set_main_option("sqlalchemy.url", database_url)
target_metadata = Base.metadata
# Base comes from the `import *` above (models.py isn't py.typed, so mypy
# can't see it) — same standard Alembic autogenerate pattern noted there.
target_metadata = Base.metadata # type: ignore[name-defined]
def run_migrations_offline() -> None:
+7
View File
@@ -55,3 +55,10 @@ python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
# Test fixtures (MockEnricher etc.) use the same InputType/OutputType
# class-attribute pattern real enrichers do — see the matching override
# and its rationale in flowsint-enrichers/pyproject.toml.
[[tool.mypy.overrides]]
module = "tests.*"
disable_error_code = ["valid-type", "attr-defined", "name-defined"]
@@ -5,9 +5,9 @@ This module provides utilities for serializing complex Python objects
into Neo4j-compatible primitive types, following the Single Responsibility Principle.
"""
from typing import Any, Callable, Dict, List, Optional, Type
from typing import Any, Callable, Dict, List, Optional, Type, Union
from pydantic import BaseModel, ValidationError
from pydantic import ValidationError
from flowsint_core.utils import flatten, unflatten
from flowsint_types import FlowsintType
@@ -180,7 +180,9 @@ class GraphSerializer:
@staticmethod
def graph_edge_to_neo4j_dict(
from_obj: BaseModel, to_obj: BaseModel, label: str
from_obj: Union[FlowsintType, GraphNode],
to_obj: Union[FlowsintType, GraphNode],
label: str,
) -> Dict[str, Any]:
"""Build a Neo4j relationship dict for matching nodes by type and label.
@@ -3,7 +3,7 @@ import json
import os
import time
from datetime import datetime
from typing import Any, Dict, List
from typing import Any, Dict, List, Optional, Union
from pydantic import ValidationError
@@ -13,6 +13,7 @@ from ..utils import to_json_serializable
from .enricher_base import Enricher
from .logger import Logger
from .types import FlowBranch, FlowStep
from .vault import VaultProtocol
class FlowOrchestrator(Enricher):
@@ -25,12 +26,12 @@ class FlowOrchestrator(Enricher):
sketch_id: str,
scan_id: str,
enricher_branches: List[FlowBranch],
vault=None,
vault: Optional[VaultProtocol] = None,
):
super().__init__(sketch_id, scan_id, vault=vault)
self.enricher_branches = enricher_branches
self.enrichers = {} # Map of nodeId -> enricher instance
self.execution_log_file = None # Path to the execution log file
self.enrichers: Dict[str, Enricher] = {} # Map of nodeId -> enricher instance
self.execution_log_file: Optional[str] = None # Path to the execution log file
self._create_execution_log()
self._load_enrichers()
@@ -45,7 +46,8 @@ class FlowOrchestrator(Enricher):
# Create filename with sketch_id and scan_id
filename = f"enricher_execution_{self.sketch_id}_{self.scan_id}.json"
self.execution_log_file = os.path.join(enricher_dir, filename)
log_file = os.path.join(enricher_dir, filename)
self.execution_log_file = log_file
# Serialize the enricher branches
serialized_branches = to_json_serializable(self.enricher_branches)
@@ -77,7 +79,7 @@ class FlowOrchestrator(Enricher):
initial_log["summary"]["total_steps"] = total_steps
# Save initial log file
with open(self.execution_log_file, "w", encoding="utf-8") as f:
with open(log_file, "w", encoding="utf-8") as f:
json.dump(initial_log, f, indent=2, ensure_ascii=False)
Logger.info(
@@ -94,7 +96,7 @@ class FlowOrchestrator(Enricher):
self.execution_log_file = None
def _update_execution_log(
self, step_entry: Dict[str, Any], status: str = None
self, step_entry: Optional[Dict[str, Any]] = None, status: Optional[str] = None
) -> None:
"""
Update the execution log file with a new step entry or status update.
@@ -259,9 +261,14 @@ class FlowOrchestrator(Enricher):
return results_mapping[ref_value]
return None
# No callers anywhere in the codebase — dead code. Its two return
# statements disagree (a dict in the no-inputs-resolved branch, a single
# resolved value via the last loop-iteration's input_key otherwise),
# which List[Any] never matched; typed Any here rather than guessing at
# the intended behavior of code nothing exercises.
def prepare_enricher_inputs(
self, step: FlowStep, results_mapping: Dict[str, Any], initial_values: List[str]
) -> List[Any]:
) -> Any:
"""
Prepare the inputs for an enricher based on the references and previous results.
Handles single references, lists, and direct values.
@@ -301,7 +308,12 @@ class FlowOrchestrator(Enricher):
def update_results_mapping(
self,
outputs: Dict[str, Any],
# Callers only ever pass enricher.execute()'s result through, which
# is validated as dict-or-list upstream — the body below only makes
# sense for the dict case (list membership/indexing by a string key
# would raise), a pre-existing gap not fixed here since it changes
# runtime behavior for enrichers whose outputs are list-shaped.
outputs: Union[Dict[str, Any], List[Any]],
step_outputs: Dict[str, str],
results_mapping: Dict[str, Any],
) -> None:
@@ -310,7 +322,7 @@ class FlowOrchestrator(Enricher):
"""
for output_key, output_ref in step_outputs.items():
if output_key in outputs:
results_mapping[output_ref] = outputs[output_key]
results_mapping[output_ref] = outputs[output_key] # type: ignore[index,call-overload]
@classmethod
def name(cls) -> str:
@@ -332,7 +344,13 @@ class FlowOrchestrator(Enricher):
def output_schema(cls) -> Dict[str, str]:
return {"branches": "array", "results": "dict"}
def scan(self, values: List[str]) -> Dict[str, Any]:
# Deliberately not Liskov-compatible with Enricher.scan (async, returns
# List[Dict[str, Any]]): the orchestrator returns one aggregate result
# for the whole flow run, not a per-item list, and this sync entry point
# exists for callers outside the async enricher-execution path. Not
# changing the signature here — that's a behavioral call about how
# FlowOrchestrator is actually invoked, not a typing fix.
def scan(self, values: List[str]) -> Dict[str, Any]: # type: ignore[override]
"""
Synchronous implementation of scan that runs the async version in an event loop
"""
@@ -351,13 +369,17 @@ class FlowOrchestrator(Enricher):
# Update execution log to indicate scan has started
self._update_execution_log(None, "running")
results = {"initial_values": values, "branches": [], "results": {}}
results: Dict[str, Any] = {
"initial_values": values,
"branches": [],
"results": {},
}
Logger.pending(self.sketch_id, {"message": "Starting enricher..."})
# Global mapping of output references to actual values
results_mapping = {}
results_mapping: Dict[str, Any] = {}
# Cache for enricher results to avoid recomputation
enricher_results_cache = {}
enricher_results_cache: Dict[str, Any] = {}
sum(len(branch.steps) for branch in self.enricher_branches)
completed_steps = 0
@@ -366,7 +388,11 @@ class FlowOrchestrator(Enricher):
for branch in self.enricher_branches:
branch_id = branch.id
branch_name = branch.name
branch_results = {"id": branch_id, "name": branch_name, "steps": []}
branch_results: Dict[str, Any] = {
"id": branch_id,
"name": branch_name,
"steps": [],
}
# Process each step in the branch
enricher_inputs = values
@@ -387,14 +413,14 @@ class FlowOrchestrator(Enricher):
enricher_name = enricher.name()
step_start_time = time.time()
step_result = {
step_result: Dict[str, Any] = {
"nodeId": node_id,
"enricher": enricher_name,
"status": "error", # Default to error, will update on success
}
# Create execution log entry
log_entry = {
log_entry: Dict[str, Any] = {
"step_id": f"{branch_id}_{node_id}",
"branch_id": branch_id,
"branch_name": branch_name,
+15
View File
@@ -54,3 +54,18 @@ python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
# Every enricher declares InputType/OutputType as plain class attributes
# (see Enricher.InputType = NotImplemented in enricher_base.py, overridden
# per-subclass) and then uses them in annotations like List[InputType] and
# reads fields off them (self.data.domain). This is the documented,
# intentional API — see flowsint-enricher-builder skill and every enricher
# in this package — not a mistake to individually silence. mypy can't
# express "this class attribute doubles as a type alias" without making
# Enricher generic (Enricher[TIn, TOut]), which would mean touching every
# enricher's class declaration and contradicts the deliberately simple
# "just set two class attributes" API. Narrowing the checks that pattern
# structurally can't satisfy, rather than ignoring per-line across ~15 files.
[[tool.mypy.overrides]]
module = "flowsint_enrichers.*"
disable_error_code = ["valid-type", "attr-defined", "name-defined"]