Skip to content

graflo.architecture.contract.ingestion.steps.models

Pydantic models for actor configuration.

Attributes

ActorConfig = Annotated[VertexActorConfig | TransformActorConfig | EdgeActorConfig | DescendActorConfig | VertexRouterActorConfig, PydanticField(discriminator='type')] module-attribute

Classes

DescendActorConfig

Bases: ConfigBaseModel

Configuration for a DescendActor.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
class DescendActorConfig(ConfigBaseModel):
    """Configuration for a DescendActor."""

    type: Literal["descend"] = PydanticField(
        default="descend", description="Actor type discriminator"
    )
    key: str | None = PydanticField(default=None, description="Key to descend into")
    any_key: bool = PydanticField(default=False, description="Process all keys")
    pipeline: list[ActorConfig] = PydanticField(
        default_factory=list,
        alias="apply",
        description="Pipeline of actors to apply to nested data",
    )

    @model_validator(mode="before")
    @classmethod
    def set_type_and_normalize(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        normalized = normalize_actor_step(cast(dict[str, Any], data))
        return normalized if normalized.get("type") == "descend" else data

    @model_validator(mode="after")
    def validate_explicit_vertex_requirements(self) -> DescendActorConfig:
        return self

Attributes

any_key = PydanticField(default=False, description='Process all keys') class-attribute instance-attribute
key = PydanticField(default=None, description='Key to descend into') class-attribute instance-attribute
pipeline = PydanticField(default_factory=list, alias='apply', description='Pipeline of actors to apply to nested data') class-attribute instance-attribute
type = PydanticField(default='descend', description='Actor type discriminator') class-attribute instance-attribute

Methods:

set_type_and_normalize(data) classmethod
Source code in graflo/architecture/contract/ingestion/steps/models.py 
836
837
838
839
840
841
842
@model_validator(mode="before")
@classmethod
def set_type_and_normalize(cls, data: Any) -> Any:
    if not isinstance(data, dict):
        return data
    normalized = normalize_actor_step(cast(dict[str, Any], data))
    return normalized if normalized.get("type") == "descend" else data
validate_explicit_vertex_requirements()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
844
845
846
@model_validator(mode="after")
def validate_explicit_vertex_requirements(self) -> DescendActorConfig:
    return self

EdgeActorConfig

Bases: EdgeEndpointMatchOptionsConfig

Configuration for an EdgeActor (logical edge + ingestion derivation; flat YAML).

Single-intent mode (default): declare source/target via from/to (static vertex type names) or source_role/target_role (slot-based dynamic resolution; source_type_field/target_type_field remain accepted aliases). One edge intent is emitted per document.

Multi-link mode (links list): declare a list of :class:EdgeLinkConfig items. Each item emits one edge intent per document, allowing a single pipeline step to produce multiple relationship types from one flat row. Mutually exclusive with all top-level source/target fields.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
class EdgeActorConfig(EdgeEndpointMatchOptionsConfig):
    """Configuration for an EdgeActor (logical edge + ingestion derivation; flat YAML).

    **Single-intent mode** (default): declare source/target via ``from``/``to`` (static
    vertex type names) or ``source_role``/``target_role`` (slot-based dynamic
    resolution; ``source_type_field``/``target_type_field`` remain accepted aliases).
    One edge intent is emitted per document.

    **Multi-link mode** (``links`` list): declare a list of :class:`EdgeLinkConfig` items.
    Each item emits one edge intent per document, allowing a single pipeline step to produce
    multiple relationship types from one flat row.  Mutually exclusive with all top-level
    source/target fields.
    """

    type: Literal["edge"] = PydanticField(
        default="edge", description="Actor type discriminator"
    )
    source: str | None = PydanticField(
        default=None,
        alias="from",
        description="Source vertex type name (optional if source_type_field/source_role is set).",
    )
    target: str | None = PydanticField(
        default=None,
        alias="to",
        description="Target vertex type name (optional if target_type_field/target_role is set).",
    )
    source_type_field: str | None = PydanticField(
        default=None,
        description=(
            "Accumulator slot segment for the source vertex (same name as the upstream "
            "VertexRouterActor role, inferred from type_field when role is omitted). EdgeActor scans "
            "acc_vertex for data at lindex.extend((source_type_field, 0)) to resolve the "
            "source type dynamically. Legacy alias for source_role."
        ),
    )
    target_type_field: str | None = PydanticField(
        default=None,
        description=(
            "Accumulator slot segment for the target vertex (same name as upstream "
            "VertexRouterActor role, inferred from type_field when role is omitted). "
            "Legacy alias for target_role."
        ),
    )
    source_role: str | None = PydanticField(
        default=None,
        description=(
            "Role slot name for the source vertex — role-first alias for source_type_field. "
            "When both are set, values must match."
        ),
    )
    target_role: str | None = PydanticField(
        default=None,
        description=(
            "Role slot name for the target vertex — role-first alias for target_type_field. "
            "When both are set, values must match."
        ),
    )
    links: list[EdgeLinkConfig] | None = PydanticField(
        default=None,
        description=(
            "Multi-intent list. When set, each item emits one edge intent per document. "
            "Mutually exclusive with all top-level source/target/role fields. "
            "Use when a single flat row encodes multiple relationships."
        ),
    )
    relation_map: dict[str, str] | None = PydanticField(
        default=None,
        description="Map raw relation values to canonical relation names.",
    )
    relation_map_only: bool = PydanticField(
        default=False,
        description=(
            "When True, a raw relation value absent from ``relation_map`` emits no "
            "edge. By default it passes through as the relation name, which is "
            "wrong for a step that must write only the mapped relations -- e.g. an "
            "inverse step, where a passed-through forward name would be written "
            "with swapped endpoints."
        ),
    )
    emit_inverse: bool = PydanticField(
        default=False,
        description=(
            "Also write the declared inverse of every edge this step writes: for "
            "``(s, t, a)`` with ``a`` paired to ``b`` in ``edge_config.inverses``, "
            "``(t, s, b)`` with the same properties. The relation is mirrored after "
            "it is resolved, so ``relation_field``, ``relation_map`` and "
            "``relation_from_key`` are all covered. Only a *materialized* inverse is "
            "written -- one whose edge ``(t, s, b)`` is declared; a pair that is "
            "only declared, or that the database maintains, stores nothing here."
        ),
    )
    strict_edge_types: bool = PydanticField(
        default=False,
        description=(
            "When True, skip documents whose resolved (source_type, target_type) pair "
            "is not pre-declared in the resource edge_config at init. "
            "When False (default), dynamic pairs are registered at runtime."
        ),
    )
    relation: str | None = PydanticField(
        default=None,
        description="Optional fixed logical relation / edge type name.",
    )
    relation_from_key: bool = PydanticField(
        default=False,
        description="Ingestion: derive per-document relation label from the location key during assembly.",
    )
    description: str | None = PydanticField(
        default=None,
        description="Optional semantic description (merged into schema Edge).",
    )
    relation_field: str | None = PydanticField(
        default=None,
        description="Ingestion: document field name for per-document relationship type.",
    )
    match_source: str | None = PydanticField(
        default=None,
        description="Ingestion: require this path segment in source locations.",
    )
    match_target: str | None = PydanticField(
        default=None,
        description="Ingestion: require this path segment in target locations.",
    )
    exclude_source: str | None = PydanticField(
        default=None,
        description="Ingestion: exclude source locations containing this segment.",
    )
    exclude_target: str | None = PydanticField(
        default=None,
        description="Ingestion: exclude target locations containing this segment.",
    )
    match: str | None = PydanticField(
        default=None,
        description="Ingestion: require this segment on both source and target locations.",
    )
    properties: list[Any] = PydanticField(
        default_factory=list,
        description="Edge properties merged into schema Edge (same forms as Edge.properties).",
    )
    vertex_weights: list[Any] = PydanticField(
        default_factory=list,
        description="Vertex-derived weight rules registered in EdgeDerivationRegistry.",
    )

    @staticmethod
    def _canonicalize_slot_key(
        role: str | None,
        legacy_type_field: str | None,
        *,
        role_name: str,
        type_field_name: str,
        context: str,
    ) -> str | None:
        """Canonicalize legacy slot-name fields to role-first semantics."""
        if (
            role is not None
            and legacy_type_field is not None
            and role != legacy_type_field
        ):
            raise ValueError(
                f"{role_name} and {type_field_name} must match when both are set in {context}."
            )
        return role if role is not None else legacy_type_field

    @model_validator(mode="after")
    def validate_type_sources(self) -> EdgeActorConfig:
        if self.links is not None:
            # Multi-link mode: top-level source/target fields must all be absent.
            has_single = any(
                [
                    self.source,
                    self.target,
                    self.source_type_field,
                    self.target_type_field,
                    self.source_role,
                    self.target_role,
                ]
            )
            if has_single:
                raise ValueError(
                    "edge 'links' is mutually exclusive with top-level "
                    "from/to/source_type_field/target_type_field/source_role/target_role."
                )
            if self.emit_inverse:
                raise ValueError(
                    "emit_inverse on an edge step with 'links' is ambiguous: set it "
                    "on each link whose inverse should be written."
                )
            return self

        # Single-intent mode: canonicalize to role-first slot names.
        # Use object.__setattr__ to bypass validate_assignment re-triggering this validator.
        source_role = self._canonicalize_slot_key(
            self.source_role,
            self.source_type_field,
            role_name="source_role",
            type_field_name="source_type_field",
            context="an edge step",
        )
        target_role = self._canonicalize_slot_key(
            self.target_role,
            self.target_type_field,
            role_name="target_role",
            type_field_name="target_type_field",
            context="an edge step",
        )
        object.__setattr__(self, "source_role", source_role)
        object.__setattr__(self, "target_role", target_role)
        object.__setattr__(self, "source_type_field", None)
        object.__setattr__(self, "target_type_field", None)

        # Each side needs exactly one of: static type or dynamic slot.
        if self.source is None and self.source_role is None:
            raise ValueError(
                "edge step requires 'from' (source), source_role, or source_type_field."
            )
        if self.target is None and self.target_role is None:
            raise ValueError(
                "edge step requires 'to' (target), target_role, or target_type_field."
            )
        if self.source is not None and self.source_role is not None:
            raise ValueError("'from' and source_type_field are mutually exclusive.")
        if self.target is not None and self.target_role is not None:
            raise ValueError("'to' and target_type_field are mutually exclusive.")
        # Mixed mode (one static + one dynamic) is valid; both-static is pure static mode.
        if self.relation_map_only and (
            self.relation_field is None or not self.relation_map
        ):
            raise ValueError(
                "relation_map_only requires relation_field and a non-empty relation_map."
            )
        return self

    @property
    def derivation(self) -> EdgeDerivation:
        """Normalized ingestion-only fields for assembly/render."""
        return EdgeDerivation(
            match_source=self.match_source,
            match_target=self.match_target,
            exclude_source=self.exclude_source,
            exclude_target=self.exclude_target,
            match=self.match,
            relation_field=self.relation_field,
            relation_from_key=self.relation_from_key,
            source_match=self.source_match,
            target_match=self.target_match,
            on_ambiguous=self.on_ambiguous,
            emit_inverse=self.emit_inverse,
        )

    @model_validator(mode="before")
    @classmethod
    def set_type_and_flatten(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        normalized = normalize_actor_step(cast(dict[str, Any], data))
        return normalized if normalized.get("type") == "edge" else data

Attributes

derivation property

Normalized ingestion-only fields for assembly/render.

description = PydanticField(default=None, description='Optional semantic description (merged into schema Edge).') class-attribute instance-attribute
emit_inverse = PydanticField(default=False, description='Also write the declared inverse of every edge this step writes: for ``(s, t, a)`` with ``a`` paired to ``b`` in ``edge_config.inverses``, ``(t, s, b)`` with the same properties. The relation is mirrored after it is resolved, so ``relation_field``, ``relation_map`` and ``relation_from_key`` are all covered. Only a *materialized* inverse is written -- one whose edge ``(t, s, b)`` is declared; a pair that is only declared, or that the database maintains, stores nothing here.') class-attribute instance-attribute
exclude_source = PydanticField(default=None, description='Ingestion: exclude source locations containing this segment.') class-attribute instance-attribute
exclude_target = PydanticField(default=None, description='Ingestion: exclude target locations containing this segment.') class-attribute instance-attribute
match = PydanticField(default=None, description='Ingestion: require this segment on both source and target locations.') class-attribute instance-attribute
match_source = PydanticField(default=None, description='Ingestion: require this path segment in source locations.') class-attribute instance-attribute
match_target = PydanticField(default=None, description='Ingestion: require this path segment in target locations.') class-attribute instance-attribute
properties = PydanticField(default_factory=list, description='Edge properties merged into schema Edge (same forms as Edge.properties).') class-attribute instance-attribute
relation = PydanticField(default=None, description='Optional fixed logical relation / edge type name.') class-attribute instance-attribute
relation_field = PydanticField(default=None, description='Ingestion: document field name for per-document relationship type.') class-attribute instance-attribute
relation_from_key = PydanticField(default=False, description='Ingestion: derive per-document relation label from the location key during assembly.') class-attribute instance-attribute
relation_map = PydanticField(default=None, description='Map raw relation values to canonical relation names.') class-attribute instance-attribute
relation_map_only = PydanticField(default=False, description='When True, a raw relation value absent from ``relation_map`` emits no edge. By default it passes through as the relation name, which is wrong for a step that must write only the mapped relations -- e.g. an inverse step, where a passed-through forward name would be written with swapped endpoints.') class-attribute instance-attribute
source = PydanticField(default=None, alias='from', description='Source vertex type name (optional if source_type_field/source_role is set).') class-attribute instance-attribute
source_role = PydanticField(default=None, description='Role slot name for the source vertex — role-first alias for source_type_field. When both are set, values must match.') class-attribute instance-attribute
source_type_field = PydanticField(default=None, description='Accumulator slot segment for the source vertex (same name as the upstream VertexRouterActor role, inferred from type_field when role is omitted). EdgeActor scans acc_vertex for data at lindex.extend((source_type_field, 0)) to resolve the source type dynamically. Legacy alias for source_role.') class-attribute instance-attribute
strict_edge_types = PydanticField(default=False, description='When True, skip documents whose resolved (source_type, target_type) pair is not pre-declared in the resource edge_config at init. When False (default), dynamic pairs are registered at runtime.') class-attribute instance-attribute
target = PydanticField(default=None, alias='to', description='Target vertex type name (optional if target_type_field/target_role is set).') class-attribute instance-attribute
target_role = PydanticField(default=None, description='Role slot name for the target vertex — role-first alias for target_type_field. When both are set, values must match.') class-attribute instance-attribute
target_type_field = PydanticField(default=None, description='Accumulator slot segment for the target vertex (same name as upstream VertexRouterActor role, inferred from type_field when role is omitted). Legacy alias for target_role.') class-attribute instance-attribute
type = PydanticField(default='edge', description='Actor type discriminator') class-attribute instance-attribute
vertex_weights = PydanticField(default_factory=list, description='Vertex-derived weight rules registered in EdgeDerivationRegistry.') class-attribute instance-attribute

Methods:

set_type_and_flatten(data) classmethod
Source code in graflo/architecture/contract/ingestion/steps/models.py 
813
814
815
816
817
818
819
@model_validator(mode="before")
@classmethod
def set_type_and_flatten(cls, data: Any) -> Any:
    if not isinstance(data, dict):
        return data
    normalized = normalize_actor_step(cast(dict[str, Any], data))
    return normalized if normalized.get("type") == "edge" else data
validate_type_sources()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
@model_validator(mode="after")
def validate_type_sources(self) -> EdgeActorConfig:
    if self.links is not None:
        # Multi-link mode: top-level source/target fields must all be absent.
        has_single = any(
            [
                self.source,
                self.target,
                self.source_type_field,
                self.target_type_field,
                self.source_role,
                self.target_role,
            ]
        )
        if has_single:
            raise ValueError(
                "edge 'links' is mutually exclusive with top-level "
                "from/to/source_type_field/target_type_field/source_role/target_role."
            )
        if self.emit_inverse:
            raise ValueError(
                "emit_inverse on an edge step with 'links' is ambiguous: set it "
                "on each link whose inverse should be written."
            )
        return self

    # Single-intent mode: canonicalize to role-first slot names.
    # Use object.__setattr__ to bypass validate_assignment re-triggering this validator.
    source_role = self._canonicalize_slot_key(
        self.source_role,
        self.source_type_field,
        role_name="source_role",
        type_field_name="source_type_field",
        context="an edge step",
    )
    target_role = self._canonicalize_slot_key(
        self.target_role,
        self.target_type_field,
        role_name="target_role",
        type_field_name="target_type_field",
        context="an edge step",
    )
    object.__setattr__(self, "source_role", source_role)
    object.__setattr__(self, "target_role", target_role)
    object.__setattr__(self, "source_type_field", None)
    object.__setattr__(self, "target_type_field", None)

    # Each side needs exactly one of: static type or dynamic slot.
    if self.source is None and self.source_role is None:
        raise ValueError(
            "edge step requires 'from' (source), source_role, or source_type_field."
        )
    if self.target is None and self.target_role is None:
        raise ValueError(
            "edge step requires 'to' (target), target_role, or target_type_field."
        )
    if self.source is not None and self.source_role is not None:
        raise ValueError("'from' and source_type_field are mutually exclusive.")
    if self.target is not None and self.target_role is not None:
        raise ValueError("'to' and target_type_field are mutually exclusive.")
    # Mixed mode (one static + one dynamic) is valid; both-static is pure static mode.
    if self.relation_map_only and (
        self.relation_field is None or not self.relation_map
    ):
        raise ValueError(
            "relation_map_only requires relation_field and a non-empty relation_map."
        )
    return self

EdgeEndpointMatchOptionsConfig

Bases: ConfigBaseModel

How an edge step locates its endpoint vertices.

By default both endpoints are matched on the vertex's primary identity, which is what every edge step has always done. Selecting a secondary identity instead lets an edge-only source reference endpoints by a business key. The choice is per endpoint, so source and target may differ.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
class EdgeEndpointMatchOptionsConfig(ConfigBaseModel):
    """How an edge step locates its endpoint vertices.

    By default both endpoints are matched on the vertex's primary ``identity``,
    which is what every edge step has always done. Selecting a secondary
    identity instead lets an edge-only source reference endpoints by a business
    key. The choice is per endpoint, so source and target may differ.
    """

    source_match: str | list[str] | None = PydanticField(
        default=None,
        description=(
            "Which identity to match the source endpoint on: omitted or 'identity' "
            "for the primary identity (default), a declared secondary identity name, "
            "an explicit field list equal to a declared secondary identity, or "
            "'secondary' when the vertex declares exactly one."
        ),
    )
    target_match: str | list[str] | None = PydanticField(
        default=None,
        description="Same as source_match, for the target endpoint.",
    )
    on_ambiguous: EndpointAmbiguityPolicy | None = PydanticField(
        default=None,
        description=(
            "Override for this step when a secondary identity matches several "
            "vertices. Inherits ingestion_model.endpoints_on_ambiguous when unset. "
            "Has no effect on endpoints matched by primary identity."
        ),
    )

Attributes

on_ambiguous = PydanticField(default=None, description='Override for this step when a secondary identity matches several vertices. Inherits ingestion_model.endpoints_on_ambiguous when unset. Has no effect on endpoints matched by primary identity.') class-attribute instance-attribute
source_match = PydanticField(default=None, description="Which identity to match the source endpoint on: omitted or 'identity' for the primary identity (default), a declared secondary identity name, an explicit field list equal to a declared secondary identity, or 'secondary' when the vertex declares exactly one.") class-attribute instance-attribute
target_match = PydanticField(default=None, description='Same as source_match, for the target endpoint.') class-attribute instance-attribute

EdgeLinkConfig

Bases: EdgeEndpointMatchOptionsConfig

One intent in a multi-link edge step.

Each item in an EdgeActorConfig.links list describes one source→target→relation binding to emit per document. Equivalent to a single-intent edge step without the links field itself.

Slot resolution uses role-first semantics (source_role / target_role). Legacy aliases (source_type_field / target_type_field) are accepted and canonicalized to their role counterparts. The slot name is the accumulator segment populated by an upstream vertex step with a matching role, or by vertex_router.role (which defaults to type_field when omitted).

Source code in graflo/architecture/contract/ingestion/steps/models.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
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
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
class EdgeLinkConfig(EdgeEndpointMatchOptionsConfig):
    """One intent in a multi-link edge step.

    Each item in an ``EdgeActorConfig.links`` list describes one source→target→relation
    binding to emit per document. Equivalent to a single-intent ``edge`` step without the
    ``links`` field itself.

    Slot resolution uses role-first semantics (``source_role`` / ``target_role``).
    Legacy aliases (``source_type_field`` / ``target_type_field``) are accepted and
    canonicalized to their role counterparts. The slot name is the accumulator segment
    populated by an upstream ``vertex`` step with a matching ``role``, or by
    ``vertex_router.role`` (which defaults to ``type_field`` when omitted).
    """

    model_config = {"extra": "forbid", "populate_by_name": True}

    source: str | None = PydanticField(
        default=None,
        alias="from",
        description="Static source vertex type name. Exclusive with source_type_field / source_role.",
    )
    target: str | None = PydanticField(
        default=None,
        alias="to",
        description="Static target vertex type name. Exclusive with target_type_field / target_role.",
    )
    source_type_field: str | None = PydanticField(
        default=None,
        description=(
            "Accumulator slot segment for the source vertex (same name as upstream "
            "vertex/vertex_router role). Exclusive with 'from' and source_role."
        ),
    )
    target_type_field: str | None = PydanticField(
        default=None,
        description=(
            "Accumulator slot segment for the target vertex (same name as upstream "
            "vertex/vertex_router role). Exclusive with 'to' and target_role."
        ),
    )
    source_role: str | None = PydanticField(
        default=None,
        description=(
            "Role-first alias for source_type_field (same accumulator segment name). "
            "When both are set, values must match."
        ),
    )
    target_role: str | None = PydanticField(
        default=None,
        description=(
            "Role-first alias for target_type_field (same accumulator segment name). "
            "When both are set, values must match."
        ),
    )
    relation: str | None = PydanticField(
        default=None,
        description="Fixed relation / edge type name for this link.",
    )
    relation_field: str | None = PydanticField(
        default=None,
        description="Document field name for per-document relationship type.",
    )
    match_source: str | None = PydanticField(
        default=None,
        description="Require this path segment in source vertex locations.",
    )
    match_target: str | None = PydanticField(
        default=None,
        description="Require this path segment in target vertex locations.",
    )
    emit_inverse: bool = PydanticField(
        default=False,
        description=(
            "Also write the declared inverse of every edge this link writes: for "
            "``(s, t, a)`` with ``a`` paired to ``b`` in ``edge_config.inverses``, "
            "``(t, s, b)`` with the same properties. The relation is mirrored after "
            "it is resolved, so ``relation_field``, ``relation_map`` and "
            "``relation_from_key`` are all covered. Only a *materialized* inverse is "
            "written -- one whose edge ``(t, s, b)`` is declared; a pair that is "
            "only declared, or that the database maintains, stores nothing here."
        ),
    )

    @staticmethod
    def _canonicalize_slot_key(
        role: str | None,
        legacy_type_field: str | None,
        *,
        role_name: str,
        type_field_name: str,
        context: str,
    ) -> str | None:
        """Canonicalize legacy slot-name fields to role-first semantics."""
        if (
            role is not None
            and legacy_type_field is not None
            and role != legacy_type_field
        ):
            raise ValueError(
                f"{role_name} and {type_field_name} must match when both are set in {context}."
            )
        return role if role is not None else legacy_type_field

    @model_validator(mode="after")
    def resolve_and_validate(self) -> EdgeLinkConfig:
        # Canonicalize to role-first slot names while preserving legacy key input.
        # Use object.__setattr__ to bypass validate_assignment re-triggering this validator.
        source_role = self._canonicalize_slot_key(
            self.source_role,
            self.source_type_field,
            role_name="source_role",
            type_field_name="source_type_field",
            context="an edge link",
        )
        target_role = self._canonicalize_slot_key(
            self.target_role,
            self.target_type_field,
            role_name="target_role",
            type_field_name="target_type_field",
            context="an edge link",
        )
        object.__setattr__(self, "source_role", source_role)
        object.__setattr__(self, "target_role", target_role)
        object.__setattr__(self, "source_type_field", None)
        object.__setattr__(self, "target_type_field", None)

        # Each side needs exactly one of: static type or slot reference.
        if self.source is None and self.source_role is None:
            raise ValueError(
                "edge link requires 'from' (source), source_role, or source_type_field."
            )
        if self.target is None and self.target_role is None:
            raise ValueError(
                "edge link requires 'to' (target), target_role, or target_type_field."
            )
        if self.source is not None and self.source_role is not None:
            raise ValueError(
                "'from' and source_type_field/source_role are mutually exclusive."
            )
        if self.target is not None and self.target_role is not None:
            raise ValueError(
                "'to' and target_type_field/target_role are mutually exclusive."
            )
        return self

Attributes

emit_inverse = PydanticField(default=False, description='Also write the declared inverse of every edge this link writes: for ``(s, t, a)`` with ``a`` paired to ``b`` in ``edge_config.inverses``, ``(t, s, b)`` with the same properties. The relation is mirrored after it is resolved, so ``relation_field``, ``relation_map`` and ``relation_from_key`` are all covered. Only a *materialized* inverse is written -- one whose edge ``(t, s, b)`` is declared; a pair that is only declared, or that the database maintains, stores nothing here.') class-attribute instance-attribute
match_source = PydanticField(default=None, description='Require this path segment in source vertex locations.') class-attribute instance-attribute
match_target = PydanticField(default=None, description='Require this path segment in target vertex locations.') class-attribute instance-attribute
model_config = {'extra': 'forbid', 'populate_by_name': True} class-attribute instance-attribute
relation = PydanticField(default=None, description='Fixed relation / edge type name for this link.') class-attribute instance-attribute
relation_field = PydanticField(default=None, description='Document field name for per-document relationship type.') class-attribute instance-attribute
source = PydanticField(default=None, alias='from', description='Static source vertex type name. Exclusive with source_type_field / source_role.') class-attribute instance-attribute
source_role = PydanticField(default=None, description='Role-first alias for source_type_field (same accumulator segment name). When both are set, values must match.') class-attribute instance-attribute
source_type_field = PydanticField(default=None, description="Accumulator slot segment for the source vertex (same name as upstream vertex/vertex_router role). Exclusive with 'from' and source_role.") class-attribute instance-attribute
target = PydanticField(default=None, alias='to', description='Static target vertex type name. Exclusive with target_type_field / target_role.') class-attribute instance-attribute
target_role = PydanticField(default=None, description='Role-first alias for target_type_field (same accumulator segment name). When both are set, values must match.') class-attribute instance-attribute
target_type_field = PydanticField(default=None, description="Accumulator slot segment for the target vertex (same name as upstream vertex/vertex_router role). Exclusive with 'to' and target_role.") class-attribute instance-attribute

Methods:

resolve_and_validate()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
@model_validator(mode="after")
def resolve_and_validate(self) -> EdgeLinkConfig:
    # Canonicalize to role-first slot names while preserving legacy key input.
    # Use object.__setattr__ to bypass validate_assignment re-triggering this validator.
    source_role = self._canonicalize_slot_key(
        self.source_role,
        self.source_type_field,
        role_name="source_role",
        type_field_name="source_type_field",
        context="an edge link",
    )
    target_role = self._canonicalize_slot_key(
        self.target_role,
        self.target_type_field,
        role_name="target_role",
        type_field_name="target_type_field",
        context="an edge link",
    )
    object.__setattr__(self, "source_role", source_role)
    object.__setattr__(self, "target_role", target_role)
    object.__setattr__(self, "source_type_field", None)
    object.__setattr__(self, "target_type_field", None)

    # Each side needs exactly one of: static type or slot reference.
    if self.source is None and self.source_role is None:
        raise ValueError(
            "edge link requires 'from' (source), source_role, or source_type_field."
        )
    if self.target is None and self.target_role is None:
        raise ValueError(
            "edge link requires 'to' (target), target_role, or target_type_field."
        )
    if self.source is not None and self.source_role is not None:
        raise ValueError(
            "'from' and source_type_field/source_role are mutually exclusive."
        )
    if self.target is not None and self.target_role is not None:
        raise ValueError(
            "'to' and target_type_field/target_role are mutually exclusive."
        )
    return self

TransformActorConfig

Bases: ConfigBaseModel

Configuration for a TransformActor.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class TransformActorConfig(ConfigBaseModel):
    """Configuration for a TransformActor."""

    type: Literal["transform"] = PydanticField(
        default="transform", description="Actor type discriminator"
    )
    when: TransformGuardConfig | None = PydanticField(
        default=None,
        description=(
            "Optional guard: the step runs only when the observation satisfies "
            "it, and writes nothing otherwise. See TransformGuardConfig."
        ),
    )
    rename: dict[str, str] | None = PydanticField(
        default=None,
        description="Rename mapping in explicit DSL form: transform.rename.",
    )
    call: TransformCallConfig | None = PydanticField(
        default=None,
        description="Function-call configuration in explicit DSL form: transform.call.",
    )

    @model_validator(mode="before")
    @classmethod
    def set_type_and_flatten(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        normalized = normalize_actor_step(cast(dict[str, Any], data))
        if normalized.get("type") != "transform":
            return data
        normalized = dict(normalized)
        call = normalized.get("call")
        if isinstance(call, dict):
            call = dict(call)
            for key in ("input", "output"):
                value = call.get(key)
                if isinstance(value, str):
                    call[key] = [value]
                elif isinstance(value, tuple):
                    call[key] = list(value)
            keys = call.get("keys")
            if isinstance(keys, str):
                call["keys"] = {"mode": "include", "names": [keys]}
            elif isinstance(keys, tuple):
                call["keys"] = {"mode": "include", "names": list(keys)}
            elif isinstance(keys, list):
                call["keys"] = {"mode": "include", "names": keys}
            elif isinstance(keys, dict):
                keys = dict(keys)
                names = keys.get("names")
                if isinstance(names, str):
                    keys["names"] = [names]
                elif isinstance(names, tuple):
                    keys["names"] = list(names)
                call["keys"] = keys
            normalized["call"] = call
        return normalized

    @model_validator(mode="after")
    def validate_mode(self) -> TransformActorConfig:
        enabled = sum([self.rename is not None, self.call is not None])
        if enabled != 1:
            raise ValueError(
                "Transform step must define exactly one of rename or call."
            )
        return self

Attributes

call = PydanticField(default=None, description='Function-call configuration in explicit DSL form: transform.call.') class-attribute instance-attribute
rename = PydanticField(default=None, description='Rename mapping in explicit DSL form: transform.rename.') class-attribute instance-attribute
type = PydanticField(default='transform', description='Actor type discriminator') class-attribute instance-attribute
when = PydanticField(default=None, description='Optional guard: the step runs only when the observation satisfies it, and writes nothing otherwise. See TransformGuardConfig.') class-attribute instance-attribute

Methods:

set_type_and_flatten(data) classmethod
Source code in graflo/architecture/contract/ingestion/steps/models.py 
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
156
157
158
159
160
161
162
163
164
@model_validator(mode="before")
@classmethod
def set_type_and_flatten(cls, data: Any) -> Any:
    if not isinstance(data, dict):
        return data
    normalized = normalize_actor_step(cast(dict[str, Any], data))
    if normalized.get("type") != "transform":
        return data
    normalized = dict(normalized)
    call = normalized.get("call")
    if isinstance(call, dict):
        call = dict(call)
        for key in ("input", "output"):
            value = call.get(key)
            if isinstance(value, str):
                call[key] = [value]
            elif isinstance(value, tuple):
                call[key] = list(value)
        keys = call.get("keys")
        if isinstance(keys, str):
            call["keys"] = {"mode": "include", "names": [keys]}
        elif isinstance(keys, tuple):
            call["keys"] = {"mode": "include", "names": list(keys)}
        elif isinstance(keys, list):
            call["keys"] = {"mode": "include", "names": keys}
        elif isinstance(keys, dict):
            keys = dict(keys)
            names = keys.get("names")
            if isinstance(names, str):
                keys["names"] = [names]
            elif isinstance(names, tuple):
                keys["names"] = list(names)
            call["keys"] = keys
        normalized["call"] = call
    return normalized
validate_mode()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
166
167
168
169
170
171
172
173
@model_validator(mode="after")
def validate_mode(self) -> TransformActorConfig:
    enabled = sum([self.rename is not None, self.call is not None])
    if enabled != 1:
        raise ValueError(
            "Transform step must define exactly one of rename or call."
        )
    return self

TransformCallConfig

Bases: ConfigBaseModel

Explicit function call transform DSL payload.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
class TransformCallConfig(ConfigBaseModel):
    """Explicit function call transform DSL payload."""

    class KeySelectionConfig(ConfigBaseModel):
        """Selection of document keys for key-target transforms."""

        mode: Literal["all", "include", "exclude"] = PydanticField(
            default="all",
            description=(
                "How keys are selected when target='keys': "
                "all=all keys, include=only provided keys, "
                "exclude=all except provided keys."
            ),
        )
        names: list[str] = PydanticField(
            default_factory=list,
            description="Key names used by include/exclude selection modes.",
        )

        @model_validator(mode="after")
        def validate_mode_names(self) -> TransformCallConfig.KeySelectionConfig:
            if self.mode == "all" and self.names:
                raise ValueError(
                    "call.keys.names must be empty when call.keys.mode='all'."
                )
            if self.mode in {"include", "exclude"} and not self.names:
                raise ValueError(
                    "call.keys.names must be provided when call.keys.mode is include/exclude."
                )
            return self

    use: str | None = PydanticField(
        default=None,
        description=(
            "Named transform reference from ingestion_model.transforms. "
            "When provided, module/foo must be omitted."
        ),
    )
    module: str | None = PydanticField(
        default=None, description="Module containing transform function."
    )
    foo: str | None = PydanticField(
        default=None, description="Transform function name in module."
    )
    params: dict[str, Any] = PydanticField(
        default_factory=dict, description="Function call keyword arguments."
    )
    input: list[str] | None = PydanticField(
        default=None, description="Input field names for function execution."
    )
    output: list[str] | None = PydanticField(
        default=None, description="Optional output field names."
    )
    input_groups: list[list[str]] | None = PydanticField(
        default=None,
        description=(
            "Explicit groups of input fields for repeated tuple-style function calls."
        ),
    )
    output_groups: list[list[str]] | None = PydanticField(
        default=None,
        description="Optional output field groups aligned with input_groups.",
    )
    target: Literal["values", "keys"] | None = PydanticField(
        default=None,
        description=(
            "Transform target. Omit with call.use to inherit from ingestion_model.transforms "
            "entry. values=transform input values, keys=transform selected document keys. "
            "Inline calls (no use) default to values when omitted."
        ),
    )
    keys: KeySelectionConfig | None = PydanticField(
        default=None,
        description="Optional key selection for target='keys'.",
    )
    strategy: Literal["single", "each", "all"] | None = PydanticField(
        default=None, description="Execution strategy for function calls."
    )
    dress: DressConfig | None = PydanticField(
        default=None, description="Pivot dressing output for scalar call results."
    )

    @model_validator(mode="before")
    @classmethod
    def normalize_io(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        data = dict(data)
        for key in ("input", "output"):
            value = data.get(key)
            if isinstance(value, str):
                data[key] = [value]
            elif isinstance(value, tuple):
                data[key] = list(value)
        for key in ("input_groups", "output_groups"):
            value = data.get(key)
            if value is None:
                continue
            if isinstance(value, tuple):
                value = list(value)
            if isinstance(value, list):
                normalized_groups: list[Any] = []
                for group in value:
                    if isinstance(group, str):
                        normalized_groups.append([group])
                    elif isinstance(group, tuple):
                        normalized_groups.append(list(group))
                    else:
                        normalized_groups.append(group)
                data[key] = normalized_groups
        keys = data.get("keys")
        if isinstance(keys, str):
            data["keys"] = {"mode": "include", "names": [keys]}
        elif isinstance(keys, tuple):
            data["keys"] = {"mode": "include", "names": list(keys)}
        elif isinstance(keys, list):
            data["keys"] = {"mode": "include", "names": keys}
        elif isinstance(keys, dict):
            keys = dict(keys)
            names = keys.get("names")
            if isinstance(names, str):
                keys["names"] = [names]
            elif isinstance(names, tuple):
                keys["names"] = list(names)
            data["keys"] = keys
        return data

    @model_validator(mode="after")
    def validate_target(self) -> TransformCallConfig:
        if self.use is not None and (self.module is not None or self.foo is not None):
            raise ValueError("call.use cannot be combined with call.module/call.foo.")
        if self.use is None:
            has_inline_callable = self.module is not None or self.foo is not None
            if has_inline_callable and (self.module is None or self.foo is None):
                raise ValueError(
                    "Inline call functions require both call.module and call.foo."
                )
            if not has_inline_callable and self.dress is None:
                raise ValueError(
                    "call must provide either use, both module+foo, or dress shorthand."
                )
            if self.dress is not None and not self.input:
                raise ValueError("dress shorthand requires call.input.")
        if self.use is None:
            effective_target: Literal["values", "keys"] | None = (
                self.target if self.target is not None else "values"
            )
        else:
            effective_target = self.target

        if effective_target == "keys":
            if self.strategy is not None:
                raise ValueError(
                    "call.strategy is not allowed when call.target='keys'; key mode uses implicit per-key execution."
                )
            if self.input:
                raise ValueError(
                    "call.input is not allowed when call.target='keys'; use call.keys selection instead."
                )
            if self.output:
                raise ValueError("call.output is not allowed when call.target='keys'.")
            if self.input_groups:
                raise ValueError(
                    "call.input_groups is not allowed when call.target='keys'."
                )
            if self.output_groups:
                raise ValueError(
                    "call.output_groups is not allowed when call.target='keys'."
                )
            if self.dress is not None:
                raise ValueError("call.dress is not supported when call.target='keys'.")
        elif effective_target == "values" and self.keys is not None:
            raise ValueError("call.keys can only be used when call.target='keys'.")
        if self.input is not None and self.input_groups is not None:
            raise ValueError(
                "Provide either call.input or call.input_groups, not both."
            )
        if self.output_groups is not None and self.input_groups is None:
            raise ValueError("call.output_groups requires call.input_groups.")
        if self.input_groups is not None:
            if self.strategy not in {None, "single"}:
                raise ValueError(
                    "call.input_groups does not support call.strategy. "
                    "Grouped mode is an explicit repeated execution."
                )
            if self.dress is not None:
                raise ValueError("call.input_groups is not compatible with call.dress.")
            if self.output is not None and self.output_groups is not None:
                raise ValueError(
                    "Provide either call.output or call.output_groups for grouped calls, not both."
                )
            if self.output_groups is not None and len(self.output_groups) != len(
                self.input_groups
            ):
                raise ValueError(
                    "call.output_groups must have the same number of groups as call.input_groups."
                )
            if self.output is not None and len(self.output) != len(self.input_groups):
                raise ValueError(
                    "For grouped scalar outputs, call.output length must equal number of call.input_groups."
                )
        if self.strategy == "all" and self.input:
            raise ValueError("call.strategy='all' does not accept call.input.")
        if self.strategy == "all" and self.input_groups:
            raise ValueError("call.strategy='all' does not accept call.input_groups.")
        return self

Attributes

dress = PydanticField(default=None, description='Pivot dressing output for scalar call results.') class-attribute instance-attribute
foo = PydanticField(default=None, description='Transform function name in module.') class-attribute instance-attribute
input = PydanticField(default=None, description='Input field names for function execution.') class-attribute instance-attribute
input_groups = PydanticField(default=None, description='Explicit groups of input fields for repeated tuple-style function calls.') class-attribute instance-attribute
keys = PydanticField(default=None, description="Optional key selection for target='keys'.") class-attribute instance-attribute
module = PydanticField(default=None, description='Module containing transform function.') class-attribute instance-attribute
output = PydanticField(default=None, description='Optional output field names.') class-attribute instance-attribute
output_groups = PydanticField(default=None, description='Optional output field groups aligned with input_groups.') class-attribute instance-attribute
params = PydanticField(default_factory=dict, description='Function call keyword arguments.') class-attribute instance-attribute
strategy = PydanticField(default=None, description='Execution strategy for function calls.') class-attribute instance-attribute
target = PydanticField(default=None, description='Transform target. Omit with call.use to inherit from ingestion_model.transforms entry. values=transform input values, keys=transform selected document keys. Inline calls (no use) default to values when omitted.') class-attribute instance-attribute
use = PydanticField(default=None, description='Named transform reference from ingestion_model.transforms. When provided, module/foo must be omitted.') class-attribute instance-attribute

Classes

KeySelectionConfig

Bases: ConfigBaseModel

Selection of document keys for key-target transforms.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class KeySelectionConfig(ConfigBaseModel):
    """Selection of document keys for key-target transforms."""

    mode: Literal["all", "include", "exclude"] = PydanticField(
        default="all",
        description=(
            "How keys are selected when target='keys': "
            "all=all keys, include=only provided keys, "
            "exclude=all except provided keys."
        ),
    )
    names: list[str] = PydanticField(
        default_factory=list,
        description="Key names used by include/exclude selection modes.",
    )

    @model_validator(mode="after")
    def validate_mode_names(self) -> TransformCallConfig.KeySelectionConfig:
        if self.mode == "all" and self.names:
            raise ValueError(
                "call.keys.names must be empty when call.keys.mode='all'."
            )
        if self.mode in {"include", "exclude"} and not self.names:
            raise ValueError(
                "call.keys.names must be provided when call.keys.mode is include/exclude."
            )
        return self
Attributes
mode = PydanticField(default='all', description="How keys are selected when target='keys': all=all keys, include=only provided keys, exclude=all except provided keys.") class-attribute instance-attribute
names = PydanticField(default_factory=list, description='Key names used by include/exclude selection modes.') class-attribute instance-attribute
Methods:
validate_mode_names()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
195
196
197
198
199
200
201
202
203
204
205
@model_validator(mode="after")
def validate_mode_names(self) -> TransformCallConfig.KeySelectionConfig:
    if self.mode == "all" and self.names:
        raise ValueError(
            "call.keys.names must be empty when call.keys.mode='all'."
        )
    if self.mode in {"include", "exclude"} and not self.names:
        raise ValueError(
            "call.keys.names must be provided when call.keys.mode is include/exclude."
        )
    return self

Methods:

normalize_io(data) classmethod
Source code in graflo/architecture/contract/ingestion/steps/models.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
@model_validator(mode="before")
@classmethod
def normalize_io(cls, data: Any) -> Any:
    if not isinstance(data, dict):
        return data
    data = dict(data)
    for key in ("input", "output"):
        value = data.get(key)
        if isinstance(value, str):
            data[key] = [value]
        elif isinstance(value, tuple):
            data[key] = list(value)
    for key in ("input_groups", "output_groups"):
        value = data.get(key)
        if value is None:
            continue
        if isinstance(value, tuple):
            value = list(value)
        if isinstance(value, list):
            normalized_groups: list[Any] = []
            for group in value:
                if isinstance(group, str):
                    normalized_groups.append([group])
                elif isinstance(group, tuple):
                    normalized_groups.append(list(group))
                else:
                    normalized_groups.append(group)
            data[key] = normalized_groups
    keys = data.get("keys")
    if isinstance(keys, str):
        data["keys"] = {"mode": "include", "names": [keys]}
    elif isinstance(keys, tuple):
        data["keys"] = {"mode": "include", "names": list(keys)}
    elif isinstance(keys, list):
        data["keys"] = {"mode": "include", "names": keys}
    elif isinstance(keys, dict):
        keys = dict(keys)
        names = keys.get("names")
        if isinstance(names, str):
            keys["names"] = [names]
        elif isinstance(names, tuple):
            keys["names"] = list(names)
        data["keys"] = keys
    return data
validate_target()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
@model_validator(mode="after")
def validate_target(self) -> TransformCallConfig:
    if self.use is not None and (self.module is not None or self.foo is not None):
        raise ValueError("call.use cannot be combined with call.module/call.foo.")
    if self.use is None:
        has_inline_callable = self.module is not None or self.foo is not None
        if has_inline_callable and (self.module is None or self.foo is None):
            raise ValueError(
                "Inline call functions require both call.module and call.foo."
            )
        if not has_inline_callable and self.dress is None:
            raise ValueError(
                "call must provide either use, both module+foo, or dress shorthand."
            )
        if self.dress is not None and not self.input:
            raise ValueError("dress shorthand requires call.input.")
    if self.use is None:
        effective_target: Literal["values", "keys"] | None = (
            self.target if self.target is not None else "values"
        )
    else:
        effective_target = self.target

    if effective_target == "keys":
        if self.strategy is not None:
            raise ValueError(
                "call.strategy is not allowed when call.target='keys'; key mode uses implicit per-key execution."
            )
        if self.input:
            raise ValueError(
                "call.input is not allowed when call.target='keys'; use call.keys selection instead."
            )
        if self.output:
            raise ValueError("call.output is not allowed when call.target='keys'.")
        if self.input_groups:
            raise ValueError(
                "call.input_groups is not allowed when call.target='keys'."
            )
        if self.output_groups:
            raise ValueError(
                "call.output_groups is not allowed when call.target='keys'."
            )
        if self.dress is not None:
            raise ValueError("call.dress is not supported when call.target='keys'.")
    elif effective_target == "values" and self.keys is not None:
        raise ValueError("call.keys can only be used when call.target='keys'.")
    if self.input is not None and self.input_groups is not None:
        raise ValueError(
            "Provide either call.input or call.input_groups, not both."
        )
    if self.output_groups is not None and self.input_groups is None:
        raise ValueError("call.output_groups requires call.input_groups.")
    if self.input_groups is not None:
        if self.strategy not in {None, "single"}:
            raise ValueError(
                "call.input_groups does not support call.strategy. "
                "Grouped mode is an explicit repeated execution."
            )
        if self.dress is not None:
            raise ValueError("call.input_groups is not compatible with call.dress.")
        if self.output is not None and self.output_groups is not None:
            raise ValueError(
                "Provide either call.output or call.output_groups for grouped calls, not both."
            )
        if self.output_groups is not None and len(self.output_groups) != len(
            self.input_groups
        ):
            raise ValueError(
                "call.output_groups must have the same number of groups as call.input_groups."
            )
        if self.output is not None and len(self.output) != len(self.input_groups):
            raise ValueError(
                "For grouped scalar outputs, call.output length must equal number of call.input_groups."
            )
    if self.strategy == "all" and self.input:
        raise ValueError("call.strategy='all' does not accept call.input.")
    if self.strategy == "all" and self.input_groups:
        raise ValueError("call.strategy='all' does not accept call.input_groups.")
    return self

TransformGuardConfig

Bases: ConfigBaseModel

A guard deciding whether a transform step runs on an observation.

The step runs only when field holds one of values (exact match — the same test a vertex_router applies to its type_map keys). When it does not, the step writes nothing: no output, no None. That is what makes a guard different from a function returning None — behind a router, which merges the transform buffer into one observation dict, a later None overwrites an earlier real value, whereas a step that did not run leaves the value alone. A missing field fails the guard.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class TransformGuardConfig(ConfigBaseModel):
    """A guard deciding whether a transform step runs on an observation.

    The step runs only when ``field`` holds one of ``values`` (exact match —
    the same test a ``vertex_router`` applies to its ``type_map`` keys). When
    it does not, the step **writes nothing**: no output, no ``None``. That is
    what makes a guard different from a function returning ``None`` — behind a
    router, which merges the transform buffer into one observation dict, a
    later ``None`` overwrites an earlier real value, whereas a step that did
    not run leaves the value alone. A missing ``field`` fails the guard.
    """

    field: str = PydanticField(
        ...,
        description="Observation field the guard reads (raw document key).",
    )
    values: list[str] = PydanticField(
        ...,
        alias="in",
        min_length=1,
        description="Values of ``field`` that let the step run; exact match.",
    )

    def passes(self, observation: Any) -> bool:
        """Whether *observation* satisfies the guard."""
        if not isinstance(observation, dict):
            return False
        return observation.get(self.field) in self.values

Attributes

field = PydanticField(..., description='Observation field the guard reads (raw document key).') class-attribute instance-attribute
values = PydanticField(..., alias='in', min_length=1, description='Values of ``field`` that let the step run; exact match.') class-attribute instance-attribute

Methods:

passes(observation)

Whether observation satisfies the guard.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
101
102
103
104
105
def passes(self, observation: Any) -> bool:
    """Whether *observation* satisfies the guard."""
    if not isinstance(observation, dict):
        return False
    return observation.get(self.field) in self.values

VertexActorConfig

Bases: VertexExtractionOptionsConfig

Configuration for a VertexActor.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class VertexActorConfig(VertexExtractionOptionsConfig):
    """Configuration for a VertexActor."""

    type: Literal["vertex"] = PydanticField(
        default="vertex", description="Actor type discriminator"
    )
    vertex: VertexName = PydanticField(
        ..., description="Name of the vertex type to create"
    )
    lookup_only: bool = PydanticField(
        default=False,
        description=(
            "When true the extracted documents are used to locate existing vertices "
            "for edge endpoints but are never written. Set this on edge-only "
            "resources, which reference a vertex without owning it — their documents "
            "typically carry a secondary identity and no primary key, so upserting "
            "them would create keyless duplicates."
        ),
    )

    @model_validator(mode="before")
    @classmethod
    def set_type(cls, data: Any) -> Any:
        if isinstance(data, dict) and "vertex" in data and "type" not in data:
            data = dict(data)
            data["type"] = "vertex"
        return data

Attributes

lookup_only = PydanticField(default=False, description='When true the extracted documents are used to locate existing vertices for edge endpoints but are never written. Set this on edge-only resources, which reference a vertex without owning it — their documents typically carry a secondary identity and no primary key, so upserting them would create keyless duplicates.') class-attribute instance-attribute
type = PydanticField(default='vertex', description='Actor type discriminator') class-attribute instance-attribute
vertex = PydanticField(..., description='Name of the vertex type to create') class-attribute instance-attribute

Methods:

set_type(data) classmethod
Source code in graflo/architecture/contract/ingestion/steps/models.py 
69
70
71
72
73
74
75
@model_validator(mode="before")
@classmethod
def set_type(cls, data: Any) -> Any:
    if isinstance(data, dict) and "vertex" in data and "type" not in data:
        data = dict(data)
        data["type"] = "vertex"
    return data

VertexExtractionOptionsConfig

Bases: ConfigBaseModel

Shared field-extraction options for vertex-like actors.

Source code in graflo/architecture/contract/ingestion/steps/models.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class VertexExtractionOptionsConfig(ConfigBaseModel):
    """Shared field-extraction options for vertex-like actors."""

    from_doc: dict[str, str] | None = PydanticField(
        default=None,
        alias="from",
        description="Projection: {vertex_field: doc_field}.",
    )
    keep_fields: list[str] | None = PydanticField(
        default=None, description="Optional list of fields to keep"
    )
    extraction_scope: Literal["full", "mapped_only"] = PydanticField(
        default="full",
        description=(
            "Field extraction policy. full (default) includes passthrough for remaining "
            "schema properties from the merged observation "
            "(doc + same-location transform buffer), while mapped_only limits extraction "
            "to explicit field mappings declared in from."
        ),
    )
    role: str | None = PydanticField(
        default=None,
        description=(
            "Optional accumulator slot segment used for storage/addressing. "
            "Vertex-like actors store observations at lindex.extend((role, 0)) when set. "
            "When omitted, actor-specific defaults may apply."
        ),
    )

Attributes

extraction_scope = PydanticField(default='full', description='Field extraction policy. full (default) includes passthrough for remaining schema properties from the merged observation (doc + same-location transform buffer), while mapped_only limits extraction to explicit field mappings declared in from.') class-attribute instance-attribute
from_doc = PydanticField(default=None, alias='from', description='Projection: {vertex_field: doc_field}.') class-attribute instance-attribute
keep_fields = PydanticField(default=None, description='Optional list of fields to keep') class-attribute instance-attribute
role = PydanticField(default=None, description='Optional accumulator slot segment used for storage/addressing. Vertex-like actors store observations at lindex.extend((role, 0)) when set. When omitted, actor-specific defaults may apply.') class-attribute instance-attribute

VertexRouterActorConfig

Bases: VertexExtractionOptionsConfig

Configuration for a VertexRouterActor.

Field handling matches :class:VertexActorConfig: optional router-level from / from_doc (and per-type vertex_from_map), optional keep_fields, and the same merged observation dict is passed to the lazily created :class:VertexActor (no separate slice / rename layer).

Source code in graflo/architecture/contract/ingestion/steps/models.py 
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
class VertexRouterActorConfig(VertexExtractionOptionsConfig):
    """Configuration for a VertexRouterActor.

    Field handling matches :class:`VertexActorConfig`: optional router-level ``from`` /
    ``from_doc`` (and per-type ``vertex_from_map``), optional ``keep_fields``, and the
    same merged observation dict is passed to the lazily created :class:`VertexActor`
    (no separate slice / rename layer).
    """

    type: Literal["vertex_router"] = PydanticField(
        default="vertex_router", description="Actor type discriminator"
    )
    type_field: str = PydanticField(
        ...,
        description=(
            "Key on the merged observation (document + same-location transform buffer) "
            "whose value determines the target vertex type (after type_map). "
            "This is a discriminator field, not the internal slot key. Use the "
            "actual column name (e.g. ``s__class_name`` or ``p_kind``)."
        ),
    )
    type_map: dict[str, str] | None = PydanticField(
        default=None,
        description="Map raw document values to vertex type names.",
    )
    vertex_from_map: dict[str, dict[str, str]] | None = PydanticField(
        default=None,
        description="Per-vertex-type field projection.",
    )

    @model_validator(mode="before")
    @classmethod
    def set_type(cls, data: Any) -> Any:
        if isinstance(data, dict) and "type_field" in data and "type" not in data:
            data = dict(data)
            data["type"] = "vertex_router"
        return data

    @model_validator(mode="after")
    def normalize_role(self) -> VertexRouterActorConfig:
        if self.role is None:
            object.__setattr__(self, "role", self.type_field)
        return self

Attributes

type = PydanticField(default='vertex_router', description='Actor type discriminator') class-attribute instance-attribute
type_field = PydanticField(..., description='Key on the merged observation (document + same-location transform buffer) whose value determines the target vertex type (after type_map). This is a discriminator field, not the internal slot key. Use the actual column name (e.g. ``s__class_name`` or ``p_kind``).') class-attribute instance-attribute
type_map = PydanticField(default=None, description='Map raw document values to vertex type names.') class-attribute instance-attribute
vertex_from_map = PydanticField(default=None, description='Per-vertex-type field projection.') class-attribute instance-attribute

Methods:

normalize_role()
Source code in graflo/architecture/contract/ingestion/steps/models.py 
887
888
889
890
891
@model_validator(mode="after")
def normalize_role(self) -> VertexRouterActorConfig:
    if self.role is None:
        object.__setattr__(self, "role", self.type_field)
    return self
set_type(data) classmethod
Source code in graflo/architecture/contract/ingestion/steps/models.py 
879
880
881
882
883
884
885
@model_validator(mode="before")
@classmethod
def set_type(cls, data: Any) -> Any:
    if isinstance(data, dict) and "type_field" in data and "type" not in data:
        data = dict(data)
        data["type"] = "vertex_router"
    return data

Functions: