symfonic.devtools.integrations

integrations

Integration-adapter boundary checks (plan task T4.2.1).

Reads the repository-owned optional-import map and enforces the static half of the integration-adapter standard over src/symfonic:

• IA-1 every module that reaches an optional distribution is registered;
• IA-2 the registered roots are the ones the module imports;
• IA-3 the registered import style is the one the source implements;
• IA-4 a missing extra is named as symfonic-core[<extra>];
• IA-5 no package __init__ imports an adapter at module level.

Run it with python -m symfonic.devtools.integrations.

AdapterEntry dataclass

AdapterEntry(module: str, family: str, kind: AdapterKind, extras: tuple[str, ...], roots: tuple[str, ...], import_style: ImportStyle, port: str, suite: str, extra_hint: str = 'self', notes: str = '')

One registered module that reaches an optional distribution.

IntegrationMap dataclass

IntegrationMap(entries: tuple[AdapterEntry, ...], optional_roots: dict[str, tuple[str, ...]], unpinned_roots: dict[str, str] = dict(), adopter_roots: dict[str, str] = dict(), base_transitive: tuple[str, ...] = (), clean_packages: tuple[str, ...] = (), integration_packages: tuple[str, ...] = ())

The optional-import map.

entries registers every module that reaches an optional distribution; optional_roots names every import root the packaging extras provide (root -> the extras that ship it), which is what makes IA-1 a completeness rule rather than a spot check; base_transitive lists roots that arrive with the base dependencies and therefore cannot be treated as optional however optional they look.

governed_roots property

governed_roots: frozenset[str]

Every root the map governs: optional, unpinned, or adopter-owned.

install_hints

install_hints(entry: AdapterEntry) -> tuple[str, ...]

The strings a caller must be able to read when the extra is gone.

Source code in src/symfonic/devtools/integrations/model.py

def install_hints(self, entry: AdapterEntry) -> tuple[str, ...]:
    """The strings a caller must be able to read when the extra is gone."""
    hints = [f"symfonic-core[{extra}]" for extra in entry.extras]
    hints += [
        self.unpinned_roots[root]
        for root in entry.roots
        if root in self.unpinned_roots
    ]
    return tuple(hints)

owning_package

owning_package(module: str) -> str

The declared package a module's import cost is attributed to.

Source code in src/symfonic/devtools/integrations/model.py

def owning_package(self, module: str) -> str:
    """The declared package a module's import cost is attributed to."""
    best = ""
    for package in (*self.integration_packages, *self.clean_packages):
        covers = module == package or module.startswith(package + ".")
        if covers and len(package) > len(best):
            best = package
    return best

root_of

root_of(target: str) -> str | None

The governed root an import target belongs to, if any.

Dotted roots (langgraph.checkpoint.postgres) live in separate optional distributions under a package that is a base dependency, so the longest declared root wins.

Source code in src/symfonic/devtools/integrations/model.py

def root_of(self, target: str) -> str | None:
    """The governed root an import target belongs to, if any.

    Dotted roots (``langgraph.checkpoint.postgres``) live in separate
    optional distributions under a package that is a base dependency,
    so the longest declared root wins.
    """
    best: str | None = None
    for root in self.governed_roots:
        covers = target == root or target.startswith(root + ".")
        if covers and (best is None or len(root) > len(best)):
            best = root
    return best

MapError

Bases: ValueError

The optional-import map is missing or structurally invalid.

carrier_modules

carrier_modules(modules: dict[str, ModuleInfo], integration_map: IntegrationMap) -> dict[str, dict[str, bool]]

Every scanned module that reaches an optional root, with its roots.

Source code in src/symfonic/devtools/integrations/introspect.py

def carrier_modules(
    modules: dict[str, ModuleInfo], integration_map: IntegrationMap
) -> dict[str, dict[str, bool]]:
    """Every scanned module that reaches an optional root, with its roots."""
    carriers: dict[str, dict[str, bool]] = {}
    for name, info in modules.items():
        roots = observed_roots(info, integration_map)
        if roots:
            carriers[name] = roots
    return carriers

collect_violations

collect_violations(modules: dict[str, ModuleInfo], integration_map: IntegrationMap) -> list[Violation]

Run every static rule and return the violations, rule-ordered.

Source code in src/symfonic/devtools/integrations/checks.py

def collect_violations(
    modules: dict[str, ModuleInfo], integration_map: IntegrationMap
) -> list[Violation]:
    """Run every static rule and return the violations, rule-ordered."""
    violations: list[Violation] = []
    violations.extend(check_registration(modules, integration_map))
    violations.extend(check_declared_roots(modules, integration_map))
    violations.extend(check_import_style(modules, integration_map))
    violations.extend(check_install_hint(modules, integration_map))
    violations.extend(check_no_eager_package_import(modules, integration_map))
    violations.sort(key=lambda violation: (violation.rule, violation.modules))
    return violations

load_map

load_map(map_path: Path) -> IntegrationMap

Parse and validate the map file into an :class:IntegrationMap.

Source code in src/symfonic/devtools/integrations/catalog.py

def load_map(map_path: Path) -> IntegrationMap:
    """Parse and validate the map file into an :class:`IntegrationMap`."""
    if not map_path.is_file():
        raise MapError(f"optional-import map not found: {map_path}")
    with map_path.open("rb") as handle:
        raw = tomllib.load(handle)
    rows = raw.get("adapter")
    if not isinstance(rows, list) or not rows:
        raise MapError(f"{map_path} declares no [[adapter]] entries")
    entries = tuple(_entry(row, map_path) for row in rows)
    roots = _roots(raw, map_path)
    base = raw.get("base", {})
    if not isinstance(base, dict):
        raise MapError(f"{map_path}: [base] must be a table")
    try:
        integration_map = IntegrationMap(
            entries=entries,
            optional_roots=roots,
            unpinned_roots=_string_table(raw, "unpinned_roots", map_path),
            adopter_roots=_string_table(raw, "adopter_roots", map_path),
            base_transitive=_texts(base, "transitive_allowed", map_path),
            clean_packages=_texts(base, "clean_packages", map_path),
            integration_packages=_texts(base, "integration_packages", map_path),
        )
    except MapError:
        raise
    except ValueError as exc:
        raise MapError(str(exc)) from exc
    _validate(integration_map, map_path)
    return integration_map

observed_roots

observed_roots(info: ModuleInfo, integration_map: IntegrationMap) -> dict[str, bool]

Optional roots this module imports -> True when at module level.

Type-checking-only imports are ignored: they cost nothing at runtime, which is the only thing the optional-dependency contract is about.

Source code in src/symfonic/devtools/integrations/introspect.py

def observed_roots(info: ModuleInfo, integration_map: IntegrationMap) -> dict[str, bool]:
    """Optional roots this module imports -> ``True`` when at module level.

    Type-checking-only imports are ignored: they cost nothing at runtime,
    which is the only thing the optional-dependency contract is about.
    """
    out: dict[str, bool] = {}
    for record in info.imports:
        if record.type_checking:
            continue
        root = integration_map.root_of(record.target)
        if root is None:
            continue
        module_level = not record.lazy
        out[root] = out.get(root, False) or module_level
    return out

observed_style

observed_style(info: ModuleInfo, integration_map: IntegrationMap) -> str

lazy / module-guarded / module-required for one module.

Source code in src/symfonic/devtools/integrations/introspect.py

def observed_style(
    info: ModuleInfo, integration_map: IntegrationMap
) -> str:
    """``lazy`` / ``module-guarded`` / ``module-required`` for one module."""
    module_level = [
        record
        for record in info.imports
        if not record.lazy
        and not record.type_checking
        and integration_map.root_of(record.target) is not None
    ]
    if not module_level:
        return "lazy"
    guarded = guarded_import_lines(info.path)
    if all(record.lineno in guarded for record in module_level):
        return "module-guarded"
    return "module-required"