Skip to content

graflo.architecture.profile.inverses

Audit of declared inverses across the whole manifest, and its inverses profile.

:mod:graflo.architecture.schema.inverse_realization says how each declared pair is realized in the schema. That is half the picture: a materialized inverse is an edge somebody has to feed, and whether every resource that writes the forward relation also feeds the inverse is a question about pipelines. :func:audit_inverses answers both halves in one report.

The split that matters in the findings is between what one side merely under-reports and what two sides contradict:

repairable The manifest states a fact in one place and omits it in another -- the inverse edge exists but a resource writing the forward relation feeds nothing into it; one mirror declares a property the other lacks; a relation is undirected everywhere but not declared symmetric. Propagating the fact cannot change meaning, so a planner may do it and say what it did. conflict The two places disagree -- a pair read from the same side, mirrors keyed differently, a native inverse whose reverse name is taken. Only the author knows which is right; these are listed and never touched.

Nothing here raises, and nothing needs the manifest to have been through finish_init: a manifest produced by a merge that no longer loads is exactly the one worth auditing. A schema runs its cross-checks while it is validated, so such a manifest cannot be built the usual way; :func:manifest_for_audit builds it block by block instead.

Attributes

Feeding = Literal['emit_inverse', 'step', 'inference', 'none'] module-attribute

How one resource feeds a materialized inverse: by mirroring its forward steps, by a step of its own, by edge inference, or not at all.

INVERSES_PROFILE = Profile(name='inverses', version=PROFILE_VERSION, assertions=(Assertion(_CONSISTENT, 'Declared inverses are realized without contradiction', True, check_inverses_consistent), Assertion(_COMPLETE, 'Every realization of an inverse is complete', True, check_inverses_complete))) module-attribute

PROFILE_VERSION = '1' module-attribute

__all__ = ['INVERSES_PROFILE', 'Feeding', 'InverseReport', 'PairStatus', 'audit_inverses', 'check_inverses_complete', 'check_inverses_consistent', 'manifest_for_audit'] module-attribute

Classes

InverseReport

Bases: ConfigBaseModel

Everything known about how a manifest realizes its declared inverses.

Source code in graflo/architecture/profile/inverses.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class InverseReport(ConfigBaseModel):
    """Everything known about how a manifest realizes its declared inverses."""

    pairs: list[PairStatus] = PydanticField(default_factory=list)
    symmetric: list[str] = PydanticField(default_factory=list)
    findings: list[InverseFinding] = PydanticField(default_factory=list)

    def repairable(self) -> list[InverseFinding]:
        """Findings a planner may fix by propagation."""
        return [f for f in self.findings if f.severity == "repairable"]

    def conflicts(self) -> list[InverseFinding]:
        """Findings only the author can settle."""
        return [f for f in self.findings if f.severity == "conflict"]

    def notes(self) -> list[InverseFinding]:
        return [f for f in self.findings if f.severity == "note"]

    def pair(self, relation: str) -> PairStatus | None:
        """The status of the pair that ``relation`` belongs to, either side."""
        return next(
            (p for p in self.pairs if relation in (p.relation, p.inverse)), None
        )

    def introduced_since(self, before: InverseReport) -> list[InverseFinding]:
        """Findings of this report that *before* did not have.

        ``audit_inverses(after).introduced_since(audit_inverses(before))`` is what
        a change did to the inverses, whatever produced the change.
        """
        known = {finding.key for finding in before.findings}
        return [finding for finding in self.findings if finding.key not in known]

    def to_lines(self) -> list[str]:
        """The report as text. The one renderer."""
        lines: list[str] = []
        for pair in self.pairs:
            # A mirror count only says something once the inverse is stored as edges.
            stored = pair.state in ("materialized", "partial")
            counts = f" ({pair.mirrored}/{pair.total} mirrored)" if stored else ""
            lines.append(f"{pair.relation} <-> {pair.inverse}: {pair.state}{counts}")
            for resource, feeding in sorted(pair.feeding.items()):
                lines.append(f"    fed in {resource}: {feeding}")
        for name in self.symmetric:
            lines.append(f"{name}: symmetric")
        if not self.pairs and not self.symmetric:
            lines.append("no declared inverses")
        for severity in ("conflict", "repairable", "note"):
            group = [f for f in self.findings if f.severity == severity]
            if not group:
                continue
            lines.append("")
            lines.append(f"{severity} ({len(group)}):")
            for finding in group:
                lines.append(f"  - [{finding.kind}] {finding.message}")
                for step in finding.steps:
                    lines.append(f"      at {step}")
        return lines

Attributes

findings = PydanticField(default_factory=list) class-attribute instance-attribute
pairs = PydanticField(default_factory=list) class-attribute instance-attribute
symmetric = PydanticField(default_factory=list) class-attribute instance-attribute

Methods:

conflicts()

Findings only the author can settle.

Source code in graflo/architecture/profile/inverses.py 
109
110
111
def conflicts(self) -> list[InverseFinding]:
    """Findings only the author can settle."""
    return [f for f in self.findings if f.severity == "conflict"]
introduced_since(before)

Findings of this report that before did not have.

audit_inverses(after).introduced_since(audit_inverses(before)) is what a change did to the inverses, whatever produced the change.

Source code in graflo/architecture/profile/inverses.py 
122
123
124
125
126
127
128
129
def introduced_since(self, before: InverseReport) -> list[InverseFinding]:
    """Findings of this report that *before* did not have.

    ``audit_inverses(after).introduced_since(audit_inverses(before))`` is what
    a change did to the inverses, whatever produced the change.
    """
    known = {finding.key for finding in before.findings}
    return [finding for finding in self.findings if finding.key not in known]
notes()
Source code in graflo/architecture/profile/inverses.py 
113
114
def notes(self) -> list[InverseFinding]:
    return [f for f in self.findings if f.severity == "note"]
pair(relation)

The status of the pair that relation belongs to, either side.

Source code in graflo/architecture/profile/inverses.py 
116
117
118
119
120
def pair(self, relation: str) -> PairStatus | None:
    """The status of the pair that ``relation`` belongs to, either side."""
    return next(
        (p for p in self.pairs if relation in (p.relation, p.inverse)), None
    )
repairable()

Findings a planner may fix by propagation.

Source code in graflo/architecture/profile/inverses.py 
105
106
107
def repairable(self) -> list[InverseFinding]:
    """Findings a planner may fix by propagation."""
    return [f for f in self.findings if f.severity == "repairable"]
to_lines()

The report as text. The one renderer.

Source code in graflo/architecture/profile/inverses.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def to_lines(self) -> list[str]:
    """The report as text. The one renderer."""
    lines: list[str] = []
    for pair in self.pairs:
        # A mirror count only says something once the inverse is stored as edges.
        stored = pair.state in ("materialized", "partial")
        counts = f" ({pair.mirrored}/{pair.total} mirrored)" if stored else ""
        lines.append(f"{pair.relation} <-> {pair.inverse}: {pair.state}{counts}")
        for resource, feeding in sorted(pair.feeding.items()):
            lines.append(f"    fed in {resource}: {feeding}")
    for name in self.symmetric:
        lines.append(f"{name}: symmetric")
    if not self.pairs and not self.symmetric:
        lines.append("no declared inverses")
    for severity in ("conflict", "repairable", "note"):
        group = [f for f in self.findings if f.severity == severity]
        if not group:
            continue
        lines.append("")
        lines.append(f"{severity} ({len(group)}):")
        for finding in group:
            lines.append(f"  - [{finding.kind}] {finding.message}")
            for step in finding.steps:
                lines.append(f"      at {step}")
    return lines

PairStatus

Bases: PairRealization

A declared pair across schema and ingestion.

Source code in graflo/architecture/profile/inverses.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class PairStatus(PairRealization):
    """A declared pair across schema and ingestion."""

    feeding: dict[str, Feeding] = PydanticField(
        default_factory=dict,
        description=(
            "Per resource that writes either relation of the pair: how it feeds "
            "the materialized inverse. Empty when nothing is materialized."
        ),
    )
    native_eligibility: list[NativeInverseViolation] | None = PydanticField(
        default=None,
        description=(
            "Rules that would be broken if the database maintained this pair; "
            "empty means eligible. None when the pair is already native."
        ),
    )
    native_candidate: str | None = PydanticField(
        default=None,
        description=(
            "The relation that would be listed in `native_inverses`: the one "
            "that has edges while its inverse has none."
        ),
    )

Attributes

feeding = PydanticField(default_factory=dict, description='Per resource that writes either relation of the pair: how it feeds the materialized inverse. Empty when nothing is materialized.') class-attribute instance-attribute
native_candidate = PydanticField(default=None, description='The relation that would be listed in `native_inverses`: the one that has edges while its inverse has none.') class-attribute instance-attribute
native_eligibility = PydanticField(default=None, description='Rules that would be broken if the database maintained this pair; empty means eligible. None when the pair is already native.') class-attribute instance-attribute

Functions:

audit_inverses(manifest)

Report how manifest realizes every declared inverse, and what is wrong with it.

Reads the schema, the physical profile and the ingestion model; changes nothing, raises nothing, and does not need the manifest to have loaded.

Source code in graflo/architecture/profile/inverses.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
def audit_inverses(manifest: GraphManifest) -> InverseReport:
    """Report how *manifest* realizes every declared inverse, and what is wrong with it.

    Reads the schema, the physical profile and the ingestion model; changes
    nothing, raises nothing, and does not need the manifest to have loaded.
    """
    schema = manifest.graph_schema
    if schema is None:
        return InverseReport()
    edge_config = schema.core_schema.edge_config
    profile = schema.db_profile
    vertex_names = {vertex.name for vertex in schema.core_schema.vertex_config.vertices}

    findings = schema_inverse_findings(schema)
    ingestion_findings, feeding_table = _ingestion_findings(manifest, schema)
    findings = [*findings, *ingestion_findings]

    pairs: list[PairStatus] = []
    for pair in pair_realizations(schema, findings=findings):
        candidate = _native_candidate(pair)
        eligibility = (
            None
            if pair.native_side is not None
            else native_inverse_violations(
                profile, edge_config, vertex_names, candidates=[candidate]
            )
        )
        if eligibility is not None:
            eligibility = [v for v in eligibility if v.relation == candidate]
        pairs.append(
            PairStatus(
                **pair.model_dump(),
                feeding=feeding_table.get(frozenset({pair.relation, pair.inverse}), {}),
                native_eligibility=eligibility,
                native_candidate=None if pair.native_side else candidate,
            )
        )
    return InverseReport(
        pairs=pairs, symmetric=list(edge_config.symmetric), findings=findings
    )

check_inverses_complete(context)

Nothing about an inverse is stated in one place and omitted in another.

Source code in graflo/architecture/profile/inverses.py 
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
def check_inverses_complete(context: CheckContext) -> AssertionResult:
    """Nothing about an inverse is stated in one place and omitted in another."""
    report = audit_inverses(context.manifest)
    findings = [
        Finding(
            assertion=_COMPLETE,
            status="warn",
            severity="warning",
            target=_target(finding),
            message=finding.message,
            detail={"kind": finding.kind, **finding.detail},
        )
        for finding in report.repairable()
    ] + [
        Finding(
            assertion=_COMPLETE,
            status="pass",
            severity="info",
            target=_target(finding),
            message=finding.message,
            detail={"kind": finding.kind, **finding.detail},
        )
        for finding in report.notes()
    ]
    return _assertion_result(
        _COMPLETE,
        "Every realization of an inverse is complete",
        findings,
        len(report.pairs) + len(report.symmetric),
    )

check_inverses_consistent(context)

No two places of the manifest contradict each other about an inverse.

Source code in graflo/architecture/profile/inverses.py 
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
def check_inverses_consistent(context: CheckContext) -> AssertionResult:
    """No two places of the manifest contradict each other about an inverse."""
    report = audit_inverses(context.manifest)
    findings = [
        Finding(
            assertion=_CONSISTENT,
            status="fail",
            severity="error",
            target=_target(finding),
            message=finding.message,
            detail={"kind": finding.kind, **finding.detail},
        )
        for finding in report.conflicts()
    ]
    return _assertion_result(
        _CONSISTENT,
        "Declared inverses are realized without contradiction",
        findings,
        len(report.pairs) + len(report.symmetric),
    )

manifest_for_audit(config)

The manifest config describes, built without the cross-block checks.

Each block is still validated on its own -- vertices, edges and the inverse table, the physical profile, the resources -- so what comes back is well-formed. What is skipped is everything that relates one block to another, which is exactly what an audit is there to report on. A manifest that loads normally is returned as loaded.

Raises:

Type Description
ValueError

when a block is malformed in itself.
Source code in graflo/architecture/profile/inverses.py 
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def manifest_for_audit(config: Mapping[str, Any]) -> GraphManifest:
    """The manifest *config* describes, built without the cross-block checks.

    Each block is still validated on its own -- vertices, edges and the inverse
    table, the physical profile, the resources -- so what comes back is
    well-formed. What is skipped is everything that relates one block to
    another, which is exactly what an audit is there to report on. A manifest
    that loads normally is returned as loaded.

    Raises:
        ValueError: when a block is malformed in itself.
    """
    try:
        manifest = GraphManifest.from_config(dict(config))
        manifest.finish_init()
        return manifest
    except ValueError:
        pass

    schema_block = config.get("schema", config.get("graph_schema"))
    schema: Schema | None = None
    if isinstance(schema_block, Mapping):
        graph = schema_block.get("graph", schema_block.get("core_schema")) or {}
        core = CoreSchema.model_construct(
            vertex_config=VertexConfig.model_validate(graph.get("vertex_config") or {}),
            edge_config=EdgeConfig.model_validate(graph.get("edge_config") or {}),
        )
        schema = Schema.model_construct(
            metadata=GraphMetadata.model_validate(schema_block.get("metadata") or {}),
            core_schema=core,
            db_profile=DatabaseProfile.model_validate(
                schema_block.get("db_profile") or {}
            ),
        )
    ingestion_block = config.get("ingestion_model")
    ingestion = (
        IngestionModel.model_validate(ingestion_block)
        if isinstance(ingestion_block, Mapping)
        else None
    )
    return GraphManifest.model_construct(
        graph_schema=schema, ingestion_model=ingestion, bindings=None
    )