graflo.architecture.evolution.preview_identity

How a merged class is keyed, as the merge preview reports it.

One :class:PreviewIdentity per vertex group: the funnel branches in the order a record tries them, every source that feeds each branch -- the raw fields a resource reads, the transform and guard that turn them into the branch value, the tag a local key carries -- and the member keys the new identity demotes to lookups. Everything is read from the group's effective declaration (the closed group, with the local keys merge adds itself), the same one merge lowers, so nothing here restates a rule.

Attributes

BranchKind = Literal['raw', 'derived', 'local_key'] module-attribute

IdentityMode = Literal['natural', 'funnel', 'carried'] module-attribute

Classes

BranchSource

Bases: ConfigBaseModel

One thing that feeds a branch: a resource's derivation or local key, or a member that carries a property branch.

Source code in graflo/architecture/evolution/preview_identity.py

class BranchSource(ConfigBaseModel):
    """One thing that feeds a branch: a resource's derivation or local key, or a member that carries a property branch."""

    resource: str | None = PydanticField(
        default=None, description="Union resource name; None for a property branch."
    )
    side: Side | None = PydanticField(
        default=None, description="The side the resource or member comes from."
    )
    exists: bool = PydanticField(
        default=True, description="False when no side has a resource of this name."
    )
    member: str | None = PydanticField(
        default=None,
        description="The member a member-keyed entry is for, or the member carrying a property branch.",
    )
    produces: list[str] = PydanticField(
        default_factory=list,
        description="Members of the group the resource produces on its side.",
    )
    input: list[str] = PydanticField(
        default_factory=list, description="Raw fields a derivation reads."
    )
    module: str | None = PydanticField(default=None, description="Transform module.")
    foo: str | None = PydanticField(default=None, description="Transform function.")
    params: dict[str, Any] = PydanticField(
        default_factory=dict, description="Transform keyword arguments."
    )
    field: str | None = PydanticField(
        default=None, description="Raw field a local key reads."
    )
    tag: str | None = PydanticField(
        default=None, description="The tag a local key value carries; '' is untagged."
    )
    tag_omitted: bool = PydanticField(
        default=False, description="The tag was left out and defaults to the origin."
    )
    auto: bool = PydanticField(
        default=False, description="Merge added this local key: nothing declared one."
    )
    when: dict[str, Any] | None = PydanticField(
        default=None, description="The declared guard, as written."
    )
    derive_at: list[int] | None = PydanticField(
        default=None, description="The declared descend step path for the resource."
    )
    label: str = PydanticField(
        default="", description="One-line caption: transform and guard, or tag."
    )

Attributes

auto = PydanticField(default=False, description='Merge added this local key: nothing declared one.') class-attribute instance-attribute

derive_at = PydanticField(default=None, description='The declared descend step path for the resource.') class-attribute instance-attribute

exists = PydanticField(default=True, description='False when no side has a resource of this name.') class-attribute instance-attribute

field = PydanticField(default=None, description='Raw field a local key reads.') class-attribute instance-attribute

foo = PydanticField(default=None, description='Transform function.') class-attribute instance-attribute

input = PydanticField(default_factory=list, description='Raw fields a derivation reads.') class-attribute instance-attribute

label = PydanticField(default='', description='One-line caption: transform and guard, or tag.') class-attribute instance-attribute

member = PydanticField(default=None, description='The member a member-keyed entry is for, or the member carrying a property branch.') class-attribute instance-attribute

module = PydanticField(default=None, description='Transform module.') class-attribute instance-attribute

params = PydanticField(default_factory=dict, description='Transform keyword arguments.') class-attribute instance-attribute

produces = PydanticField(default_factory=list, description='Members of the group the resource produces on its side.') class-attribute instance-attribute

resource = PydanticField(default=None, description='Union resource name; None for a property branch.') class-attribute instance-attribute

side = PydanticField(default=None, description='The side the resource or member comes from.') class-attribute instance-attribute

tag = PydanticField(default=None, description="The tag a local key value carries; '' is untagged.") class-attribute instance-attribute

tag_omitted = PydanticField(default=False, description='The tag was left out and defaults to the origin.') class-attribute instance-attribute

when = PydanticField(default=None, description='The declared guard, as written.') class-attribute instance-attribute

PreviewBranch

Bases: ConfigBaseModel

One funnel branch, or one derive attribute.

Source code in graflo/architecture/evolution/preview_identity.py

class PreviewBranch(ConfigBaseModel):
    """One funnel branch, or one ``derive`` attribute."""

    position: int = PydanticField(
        ..., description="1-based order a record tries it in; 0 for a `derive` entry."
    )
    id: str = PydanticField(..., description="Branch id, as the funnel names it.")
    kind: BranchKind = PydanticField(..., description="What the branch keys on.")
    fields: list[str] = PydanticField(
        default_factory=list, description="Merged attributes the branch reads."
    )
    sources: list[BranchSource] = PydanticField(default_factory=list)
    sep: str | None = PydanticField(
        default=None,
        description="What joins a local key's tag and value; None for other kinds.",
    )

Attributes

fields = PydanticField(default_factory=list, description='Merged attributes the branch reads.') class-attribute instance-attribute

id = PydanticField(..., description='Branch id, as the funnel names it.') class-attribute instance-attribute

kind = PydanticField(..., description='What the branch keys on.') class-attribute instance-attribute

position = PydanticField(..., description='1-based order a record tries it in; 0 for a `derive` entry.') class-attribute instance-attribute

sep = PydanticField(default=None, description="What joins a local key's tag and value; None for other kinds.") class-attribute instance-attribute

sources = PydanticField(default_factory=list) class-attribute instance-attribute

PreviewDemotion

Bases: ConfigBaseModel

A member key the new identity demotes to a lookup-only secondary.

Source code in graflo/architecture/evolution/preview_identity.py

class PreviewDemotion(ConfigBaseModel):
    """A member key the new identity demotes to a lookup-only secondary."""

    side: Side
    member: str
    fields: list[str] = PydanticField(
        default_factory=list,
        description="The key fields, as the member spells them after its vocabulary renames.",
    )
    merged_fields: list[str] = PydanticField(
        default_factory=list, description="Their names on the merged class."
    )
    key_space: str = PydanticField(..., description="What names the demoted key.")
    tags: list[str] = PydanticField(
        default_factory=list, description="Every key space the op's tags name for it."
    )

Attributes

fields = PydanticField(default_factory=list, description='The key fields, as the member spells them after its vocabulary renames.') class-attribute instance-attribute

key_space = PydanticField(..., description='What names the demoted key.') class-attribute instance-attribute

member instance-attribute

merged_fields = PydanticField(default_factory=list, description='Their names on the merged class.') class-attribute instance-attribute

side instance-attribute

tags = PydanticField(default_factory=list, description="Every key space the op's tags name for it.") class-attribute instance-attribute

PreviewIdentity

Bases: ConfigBaseModel

How a merged class is keyed, and where every key value comes from.

Source code in graflo/architecture/evolution/preview_identity.py

class PreviewIdentity(ConfigBaseModel):
    """How a merged class is keyed, and where every key value comes from."""

    mode: IdentityMode
    branches: list[PreviewBranch] = PydanticField(default_factory=list)
    derive: list[PreviewBranch] = PydanticField(
        default_factory=list,
        description="`derive` entries: attributes a branch keys on, computed per resource.",
    )
    digest_field: str | None = PydanticField(
        default=None, description="Where a funnel stores its digest."
    )
    retire: str | None = PydanticField(
        default=None, description="What happens to each member's own key."
    )
    derive_at: dict[str, list[int]] = PydanticField(default_factory=dict)
    demoted: list[PreviewDemotion] = PydanticField(default_factory=list)

    def role_of(self, attribute: str) -> list[int]:
        """Positions of the branches that key on merged *attribute*."""
        return [b.position for b in self.branches if attribute in b.fields]

Attributes

branches = PydanticField(default_factory=list) class-attribute instance-attribute

demoted = PydanticField(default_factory=list) class-attribute instance-attribute

derive = PydanticField(default_factory=list, description='`derive` entries: attributes a branch keys on, computed per resource.') class-attribute instance-attribute

derive_at = PydanticField(default_factory=dict) class-attribute instance-attribute

digest_field = PydanticField(default=None, description='Where a funnel stores its digest.') class-attribute instance-attribute

mode instance-attribute

retire = PydanticField(default=None, description="What happens to each member's own key.") class-attribute instance-attribute

Methods:

role_of(attribute)

Positions of the branches that key on merged attribute.

Source code in graflo/architecture/evolution/preview_identity.py

def role_of(self, attribute: str) -> list[int]:
    """Positions of the branches that key on merged *attribute*."""
    return [b.position for b in self.branches if attribute in b.fields]

Functions:

declared_identity(declaration)

The identity declaration states, read without the manifests.

What :func:identity_block reports minus everything only the manifests settle: no resource has a side, no member carries a property branch, an omitted tag stays unresolved, and nothing is demoted or added automatically.

Source code in graflo/architecture/evolution/preview_identity.py

def declared_identity(declaration: VertexEquivalence) -> PreviewIdentity:
    """The identity *declaration* states, read without the manifests.

    What :func:`identity_block` reports minus everything only the manifests
    settle: no resource has a side, no member carries a property branch, an
    omitted tag stays unresolved, and nothing is demoted or added
    automatically.
    """
    identity = declaration.identity
    if identity is None:
        return PreviewIdentity(mode="carried")

    def resource(name: str) -> BranchSource:
        return BranchSource(resource=name, derive_at=declaration.derive_at.get(name))

    def derived(branch: DerivedBranch, position: int) -> PreviewBranch:
        return PreviewBranch(
            position=position,
            id=branch.name,
            kind="derived",
            fields=[branch.name],
            sources=[
                resource(name).model_copy(
                    update={
                        "member": member,
                        "input": list(spec.input),
                        "module": spec.module,
                        "foo": spec.foo,
                        "params": dict(spec.params),
                        "when": _guard(spec),
                        "label": derivation_label(spec),
                    }
                )
                for name, entry in branch.sources.items()
                for member, spec in _entries(entry)
            ],
        )

    def local_key(branch: LocalKeyBranch, position: int) -> PreviewBranch:
        return PreviewBranch(
            position=position,
            id=branch.name,
            kind="local_key",
            fields=[branch.name],
            sources=[
                resource(name).model_copy(
                    update={
                        "member": member,
                        "field": source.field,
                        "tag": None if source.tag_omitted else source.tag,
                        "tag_omitted": source.tag_omitted,
                        "when": _guard(source),
                        "label": local_key_label(source, origin=None, auto=False),
                    }
                )
                for name, entry in branch.local_key.items()
                for member, source in _entries(entry)
            ],
            sep=branch.sep,
        )

    branches: list[PreviewBranch] = []
    for position, branch in enumerate(identity, start=1):
        if isinstance(branch, DerivedBranch):
            branches.append(derived(branch, position))
        elif isinstance(branch, LocalKeyBranch):
            branches.append(local_key(branch, position))
        else:
            branches.append(
                PreviewBranch(
                    position=position,
                    id=branch_id(branch),
                    kind="raw",
                    fields=branch_fields(branch),
                )
            )
    funnel = declaration.has_derivation or len(identity) > 1
    return PreviewIdentity(
        mode="funnel" if funnel else "natural",
        branches=branches,
        derive=[derived(b, 0) for b in declaration.derive_attributes()],
        digest_field=declaration.digest_field if funnel else None,
        retire=declaration.retire,
        derive_at={r: list(p) for r, p in declaration.derive_at.items()},
    )

derivation_label(spec)

normalized_key(serial_number), with params, module and guard when set.

Source code in graflo/architecture/evolution/preview_identity.py

def derivation_label(spec: DerivationSpec) -> str:
    """``normalized_key(serial_number)``, with params, module and guard when set."""
    name = spec.foo if spec.module == _DEFAULT_MODULE else f"{spec.module}.{spec.foo}"
    args = ", ".join(spec.input)
    if spec.params:
        args += "; " + ", ".join(f"{k}={v!r}" for k, v in spec.params.items())
    return f"{name}({args}){_guard_label(spec.when)}"

identity_block(cluster, resolution, sides, composite)

The :class:PreviewIdentity of one vertex group.

Parameters:

Name	Type	Description	Default
cluster	Cluster	The group, with its effective declaration.	required
resolution	ClusterResolution	The naming resolution it came from.	required
sides	Mapping[str, GraphManifest]	Both manifests, resources renamed as the union names them.	required
composite	Mapping[Side, CanonicalizeOp]	Each side's composite relabel.	required

Source code in graflo/architecture/evolution/preview_identity.py

def identity_block(
    cluster: Cluster,
    resolution: ClusterResolution,
    sides: Mapping[str, GraphManifest],
    composite: Mapping[Side, CanonicalizeOp],
) -> PreviewIdentity:
    """The :class:`PreviewIdentity` of one vertex group.

    Args:
        cluster: The group, with its effective declaration.
        resolution: The naming resolution it came from.
        sides: Both manifests, resources renamed as the union names them.
        composite: Each side's composite relabel.
    """
    from .merge import _members_produced

    declaration = cluster.declaration
    assert isinstance(declaration, VertexEquivalence)
    identity = declaration.identity
    demoted = _demotions(cluster, resolution)
    if identity is None:
        return PreviewIdentity(mode="carried", demoted=demoted)
    funnel = declaration.has_derivation or len(identity) > 1
    auto = resolution.auto_key_sources.get(cluster.into or "", frozenset())
    origins = resolution.origins

    def resource_source(resource: str) -> BranchSource:
        produced = _members_produced(resource, cluster, sides)
        if produced is None:
            return BranchSource(resource=resource, exists=False)
        side, members = produced
        return BranchSource(
            resource=resource,
            side=side,
            produces=members,
            derive_at=declaration.derive_at.get(resource),
        )

    def derived(branch: DerivedBranch, position: int) -> PreviewBranch:
        sources: list[BranchSource] = []
        for resource, entry in branch.sources.items():
            base = resource_source(resource)
            for member, spec in _entries(entry):
                sources.append(
                    base.model_copy(
                        update={
                            "member": member,
                            "input": list(spec.input),
                            "module": spec.module,
                            "foo": spec.foo,
                            "params": dict(spec.params),
                            "when": _guard(spec),
                            "label": derivation_label(spec),
                        }
                    )
                )
        return PreviewBranch(
            position=position,
            id=branch.name,
            kind="derived",
            fields=[branch.name],
            sources=sources,
        )

    def local_key(branch: LocalKeyBranch, position: int) -> PreviewBranch:
        sources: list[BranchSource] = []
        for resource, entry in branch.local_key.items():
            base = resource_source(resource)
            for member, source in _entries(entry):
                is_auto = (resource, member) in auto
                origin = origins.get(base.side, base.side) if base.side else None
                sources.append(
                    base.model_copy(
                        update={
                            "member": member,
                            "field": source.field,
                            "tag": origin if source.tag_omitted else source.tag,
                            "tag_omitted": source.tag_omitted,
                            "auto": is_auto,
                            "when": _guard(source),
                            "label": local_key_label(
                                source, origin=origin, auto=is_auto
                            ),
                        }
                    )
                )
        return PreviewBranch(
            position=position,
            id=branch.name,
            kind="local_key",
            fields=[branch.name],
            sources=sources,
            sep=branch.sep,
        )

    branches: list[PreviewBranch] = []
    for position, branch in enumerate(identity, start=1):
        if isinstance(branch, DerivedBranch):
            branches.append(derived(branch, position))
        elif isinstance(branch, LocalKeyBranch):
            branches.append(local_key(branch, position))
        else:
            fields = branch_fields(branch)
            branches.append(
                PreviewBranch(
                    position=position,
                    id=branch_id(branch),
                    kind="raw",
                    fields=fields,
                    sources=_carriers(cluster, fields, sides, composite),
                )
            )
    return PreviewIdentity(
        mode="funnel" if funnel else "natural",
        branches=branches,
        derive=[derived(b, 0) for b in declaration.derive_attributes()],
        digest_field=declaration.digest_field if funnel else None,
        retire=declaration.retire,
        derive_at={r: list(p) for r, p in declaration.derive_at.items()},
        demoted=demoted,
    )

local_key_label(source, *, origin, auto)

tag maintenance, tag left (default), untagged; · auto when merge added it.

Source code in graflo/architecture/evolution/preview_identity.py

def local_key_label(source: LocalKeySource, *, origin: str | None, auto: bool) -> str:
    """``tag maintenance``, ``tag left (default)``, ``untagged``; `` · auto`` when merge added it."""
    if source.tag_omitted:
        text = f"tag {origin} (default)" if origin else "tag (origin)"
    elif source.tag:
        text = f"tag {source.tag}"
    else:
        text = "untagged"
    if auto:
        text += " · auto"
    return text + _guard_label(source.when)