Skip to content

graflo.architecture.contract.ingestion.steps

Actor configuration models and parsing.

Modules:

Name Description
models

Pydantic models for actor configuration.
normalize

Normalization of raw actor step dicts for validation.
parse

Parsing and validation of actor configuration.
ref

Addressing edge steps inside a pipeline.

Attributes

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

__all__ = ['ActorConfig', 'DescendActorConfig', 'EdgeActorConfig', 'EdgeLinkConfig', 'TransformActorConfig', 'TransformGuardConfig', 'VertexActorConfig', 'VertexRouterActorConfig', 'canonical_actor_step', 'normalize_actor_step', 'parse_root_config', 'validate_actor_step'] 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

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

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

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:

canonical_actor_step(step)

One spelling for a pipeline step, whatever spelling it was authored in.

A step accepts several equivalent spellings -- {"vertex": v} with or without an explicit type, {"edge": {...}} or its flat body, from/to or source/target, a descend with apply or pipeline. Stored as authored, two manifests meaning the same thing compare and hash differently. This validates the step into its actor model and renders the minimal canonical dict with the discriminator made explicit.

Lists inside a step keep their order: whether any of them is a set has not been established, and preserving is the safe direction for a content hash.

A step that does not validate on its own is returned unchanged, so a canonical form never fails where the raw one would have been accepted.

Source code in graflo/architecture/contract/ingestion/steps/parse.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def canonical_actor_step(step: dict[str, Any]) -> dict[str, Any]:
    """One spelling for a pipeline step, whatever spelling it was authored in.

    A step accepts several equivalent spellings -- ``{"vertex": v}`` with or
    without an explicit ``type``, ``{"edge": {...}}`` or its flat body,
    ``from``/``to`` or ``source``/``target``, a ``descend`` with ``apply`` or
    ``pipeline``. Stored as authored, two manifests meaning the same thing
    compare and hash differently. This validates the step into its actor model
    and renders the minimal canonical dict with the discriminator made explicit.

    Lists inside a step keep their order: whether any of them is a set has not
    been established, and preserving is the safe direction for a content hash.

    A step that does not validate on its own is returned unchanged, so a
    canonical form never fails where the raw one would have been accepted.
    """
    try:
        model = _actor_config_adapter.validate_python(normalize_actor_step(step))
    except (ValidationError, ValueError, TypeError):
        return step
    return {"type": model.type, **model.to_minimal_canonical_dict()}

normalize_actor_step(data)

Normalize a raw step dict so it has 'type' and flat structure for validation.

Source code in graflo/architecture/contract/ingestion/steps/normalize.py 
 13
 14
 15
 16
 17
 18
 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
 47
 48
 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
 76
 77
 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
106
107
108
109
110
111
112
113
114
def normalize_actor_step(data: dict[str, Any]) -> dict[str, Any]:
    """Normalize a raw step dict so it has 'type' and flat structure for validation."""
    if not isinstance(data, dict):
        return data
    data = dict(data)
    if "type" in data:
        # An already-typed step still owns its children. Returning here without
        # descending left a `descend` written in its own typed form unable to
        # carry the shorthand sub-steps its `{descend: {...}}` spelling accepts,
        # because `pipeline` is a union discriminated on `type`.
        if data["type"] == "descend":
            sub = (
                data.pop("apply", None) if "pipeline" not in data else data["pipeline"]
            )
            if sub is not None:
                data["pipeline"] = [normalize_actor_step(s) for s in _steps_list(sub)]
        return data

    if "vertex" in data:
        data["type"] = "vertex"
        return data

    if "edge" in data:
        inner = data.pop("edge")
        if isinstance(inner, dict):
            data.update(inner)
        data["type"] = "edge"
        return data
    if ("source" in data or "from" in data) and ("target" in data or "to" in data):
        data = dict(data)
        data["type"] = "edge"
        return data
    if "create_edge" in data:
        inner = data.pop("create_edge")
        if isinstance(inner, dict):
            data.update(inner)
        data["type"] = "edge"
        return data

    if "descend" in data:
        inner = data.pop("descend")
        if isinstance(inner, dict):
            if "pipeline" in inner:
                inner["pipeline"] = [
                    normalize_actor_step(s) for s in _steps_list(inner["pipeline"])
                ]
            elif "apply" in inner:
                inner["pipeline"] = [
                    normalize_actor_step(s) for s in _steps_list(inner["apply"])
                ]
                del inner["apply"]
            data.update(inner)
        data["type"] = "descend"
        if "pipeline" not in data and "apply" in data:
            data["pipeline"] = [
                normalize_actor_step(s) for s in _steps_list(data["apply"])
            ]
            del data["apply"]
        return data

    if "vertex_router" in data:
        inner = data.pop("vertex_router")
        if isinstance(inner, dict):
            data.update(inner)
        data["type"] = "vertex_router"
        return data

    # Flat router form. ``VertexRouterActorConfig`` infers the discriminator from a
    # bare ``type_field`` too, so a normalizer that did not would hand every
    # structural pipeline scan (referenced vertices, renames, level lookup) a step
    # with no ``type`` — and those scans would silently skip a real router. The edge
    # branches above already claimed ``source_type_field`` / ``target_type_field``.
    if "type_field" in data:
        data["type"] = "vertex_router"
        return data

    if "transform" in data:
        inner = data.pop("transform")
        if not isinstance(inner, dict):
            raise ValueError("transform step must be an object with rename or call.")
        data.update(inner)
        data["type"] = "transform"
        return data

    if "apply" in data:
        data["type"] = "descend"
        data["pipeline"] = [normalize_actor_step(s) for s in _steps_list(data["apply"])]
        del data["apply"]
        return data
    if "pipeline" in data:
        data["type"] = "descend"
        data["pipeline"] = [
            normalize_actor_step(s) for s in _steps_list(data["pipeline"])
        ]
        return data

    if "type" not in data and ("rename" in data or "call" in data):
        data = dict(data)
        data["type"] = "transform"
        return data

    return data

parse_root_config(*args, **kwargs)

Parse root input into a single ActorConfig (single step or descend pipeline).

Source code in graflo/architecture/contract/ingestion/steps/parse.py 
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
def parse_root_config(
    *args: Any,
    **kwargs: Any,
) -> ActorConfig:
    """Parse root input into a single ActorConfig (single step or descend pipeline)."""
    pipeline: list[dict[str, Any]] | None = None
    single: dict[str, Any] | None = None

    if kwargs and ("apply" in kwargs or "pipeline" in kwargs):
        raw = kwargs.get("pipeline") or kwargs.get("apply")
        if raw is not None:
            pipeline = cast(
                list[dict[str, Any]],
                list(raw) if isinstance(raw, list) else [raw],
            )
    elif args:
        if len(args) == 1 and isinstance(args[0], list):
            list_arg = args[0]
            if not all(isinstance(item, dict) for item in list_arg):
                raise ValueError("pipeline must be a list of dict actor steps")
            pipeline = [dict(item) for item in list_arg]
        elif len(args) == 1 and isinstance(args[0], dict):
            single = dict(args[0])
        elif args and all(isinstance(a, dict) for a in args):
            pipeline = [dict(a) for a in args]

    if pipeline is not None:
        configs = []
        for step in pipeline:
            normalized = normalize_actor_step(step)
            try:
                configs.append(_actor_config_adapter.validate_python(normalized))
            except ValidationError as err:
                _raise_step_validation_error(normalized, err)
        return DescendActorConfig.model_validate(
            {
                "type": "descend",
                "key": None,
                "any_key": False,
                "pipeline": configs,
            }
        )
    if single is not None:
        step_dict = {k: v for k, v in single.items() if k not in _STEP_STRIP_KEYS}
        return validate_actor_step(normalize_actor_step(step_dict))
    step_kwargs = {k: v for k, v in kwargs.items() if k not in _STEP_STRIP_KEYS}
    return validate_actor_step(normalize_actor_step(step_kwargs))

validate_actor_step(data)

Validate a normalized step dict as ActorConfig (discriminated union).

Source code in graflo/architecture/contract/ingestion/steps/parse.py 
70
71
72
73
74
75
76
77
def validate_actor_step(
    data: dict[str, Any],
) -> ActorConfig:
    """Validate a normalized step dict as ActorConfig (discriminated union)."""
    try:
        return _actor_config_adapter.validate_python(data)
    except ValidationError as err:
        _raise_step_validation_error(data, err)