graflo.architecture.graph_types

Graph runtime types and data structures (extraction, assembly, containers).

This module defines graph-processing data structures used across the ingestion pipeline and database adapters. It provides:

• Core data types for vertices and edges
• Database index configurations
• Graph container implementations
• Edge mapping and casting utilities
• Action context for graph transformations

The module is designed to be database-agnostic, supporting both ArangoDB and Neo4j through the DBType enum. It provides a unified interface for working with graph data structures while allowing for database-specific optimizations and features.

Key Components

• EdgeMapping: Defines how edges are mapped between vertices
• IndexType: Supported database index types
• EdgeType: Types of edge handling in the graph database
• GraphContainer: Main container for graph data
• ActionContext: Context for graph transformation operations

Example

container = GraphContainer(vertices={}, edges={}, linear=[]) index = Index(fields=["name", "age"], type=IndexType.PERSISTENT) context = ActionContext()

ABCFields

Bases: ConfigBaseModel

Base model for entities that have fields.

Attributes:

Name	Type	Description
name	str | None	Optional name of the entity
fields	list[str]	List of field names

Source code in graflo/architecture/graph_types.py

class ABCFields(ConfigBaseModel):
    """Base model for entities that have fields.

    Attributes:
        name: Optional name of the entity
        fields: List of field names
    """

    name: str | None = Field(
        default=None,
        description="Optional name of the entity (e.g. vertex name for composite field prefix).",
    )
    fields: list[str] = Field(
        default_factory=list,
        description="List of field names for this entity.",
    )
    keep_vertex_name: bool = Field(
        default=True,
        description="If True, composite field names use entity@field format; otherwise use field only.",
    )

    def cfield(self, x: str) -> str:
        """Creates a composite field name by combining the entity name with a field name.

        Args:
            x: Field name to combine with entity name

        Returns:
            Composite field name in format "entity@field"
        """
        return f"{self.name}@{x}" if self.keep_vertex_name else x

cfield(x)

Creates a composite field name by combining the entity name with a field name.

Parameters:

Name	Type	Description	Default
x	str	Field name to combine with entity name	required

Returns:

Type	Description
str	Composite field name in format "entity@field"

Source code in graflo/architecture/graph_types.py

def cfield(self, x: str) -> str:
    """Creates a composite field name by combining the entity name with a field name.

    Args:
        x: Field name to combine with entity name

    Returns:
        Composite field name in format "entity@field"
    """
    return f"{self.name}@{x}" if self.keep_vertex_name else x

ActionContext

Bases: ExtractionContext

Backward-compatible extraction+assembly context.

Kept for existing callers/tests while Wave 5 migrates call surfaces.

Source code in graflo/architecture/graph_types.py

class ActionContext(ExtractionContext):
    """Backward-compatible extraction+assembly context.

    Kept for existing callers/tests while Wave 5 migrates call surfaces.
    """

    acc_global: Any = Field(default_factory=dd_factory)

AssemblyContext

Bases: ConfigBaseModel

Assembly-phase context built from extraction outputs.

Source code in graflo/architecture/graph_types.py

class AssemblyContext(ConfigBaseModel):
    """Assembly-phase context built from extraction outputs."""

    model_config = ConfigDict(kw_only=True)  # type: ignore[assignment]

    extraction: ExtractionContext
    acc_global: Any = Field(default_factory=dd_factory)

    @property
    def acc_vertex(self) -> Any:
        return self.extraction.acc_vertex

    @property
    def buffer_transforms(self) -> Any:
        return self.extraction.buffer_transforms

    @property
    def edge_requests(self) -> Any:
        return self.extraction.edge_requests

    @edge_requests.setter
    def edge_requests(self, value: Any) -> None:
        self.extraction.edge_requests = value

    @property
    def edge_intents(self) -> list[EdgeIntent]:
        return self.extraction.edge_intents

    @classmethod
    def from_extraction(cls, extraction: ExtractionContext) -> AssemblyContext:
        return cls(extraction=extraction)

EdgeCastingType

Bases: BaseEnum

Types of edge casting supported.

PAIR: Edges are cast as pairs of vertices PRODUCT: Edges are cast as combinations of vertex sets

Source code in graflo/architecture/graph_types.py

class EdgeCastingType(BaseEnum):
    """Types of edge casting supported.

    PAIR: Edges are cast as pairs of vertices
    PRODUCT: Edges are cast as combinations of vertex sets
    """

    PAIR = "pair"
    PRODUCT = "product"
    COMBINATIONS = "combinations"

EdgeIntent

Bases: ConfigBaseModel

Typed edge assembly request emitted during extraction.

Source code in graflo/architecture/graph_types.py

class EdgeIntent(ConfigBaseModel):
    """Typed edge assembly request emitted during extraction."""

    edge: Any
    location: LocationIndex | None = None
    provenance: ProvenancePath | None = None

EdgeMapping

Bases: BaseEnum

Defines how edges are mapped between vertices.

ALL: Maps all vertices to all vertices ONE_N: Maps one vertex to many vertices

Source code in graflo/architecture/graph_types.py

class EdgeMapping(BaseEnum):
    """Defines how edges are mapped between vertices.

    ALL: Maps all vertices to all vertices
    ONE_N: Maps one vertex to many vertices
    """

    ALL = "all"
    ONE_N = "1-n"

EdgeType

Bases: BaseEnum

Defines how edges are handled in the graph database.

INDIRECT: Uses pre-existing DB structures and may be used after data ingestion DIRECT: Generated during ingestion from resource pipelines

Source code in graflo/architecture/graph_types.py

class EdgeType(BaseEnum):
    """Defines how edges are handled in the graph database.

    INDIRECT: Uses pre-existing DB structures and may be used after data ingestion
    DIRECT: Generated during ingestion from resource pipelines
    """

    INDIRECT = "indirect"
    DIRECT = "direct"

EncodingType

Bases: BaseEnum

Supported character encodings for data input/output.

Source code in graflo/architecture/graph_types.py

class EncodingType(BaseEnum):
    """Supported character encodings for data input/output."""

    ISO_8859 = "ISO-8859-1"
    UTF_8 = "utf-8"

ExtractionContext

Bases: ConfigBaseModel

Extraction-phase context.

Attributes:

Name	Type	Description
acc_vertex	Any	Local accumulation of extracted vertices
buffer_transforms	Any	Buffer for transform payloads (defaultdict[LocationIndex, list])
edge_requests	Any	Explicit edge assembly requests collected during extraction
vertex_observations	list[VertexObservation]	Explicit extracted vertex observations
transform_observations	list[TransformObservation]	Explicit extracted transform observations
edge_intents	list[EdgeIntent]	Explicit edge intents for assembly phase

Source code in graflo/architecture/graph_types.py

class ExtractionContext(ConfigBaseModel):
    """Extraction-phase context.

    Attributes:
        acc_vertex: Local accumulation of extracted vertices
        buffer_transforms: Buffer for transform payloads (defaultdict[LocationIndex, list])
        edge_requests: Explicit edge assembly requests collected during extraction
        vertex_observations: Explicit extracted vertex observations
        transform_observations: Explicit extracted transform observations
        edge_intents: Explicit edge intents for assembly phase
    """

    model_config = ConfigDict(kw_only=True)  # type: ignore[assignment]

    # Pydantic cannot schema nested defaultdict with custom key types (e.g. LocationIndex),
    # so we use Any; runtime type is as documented in Attributes
    acc_vertex: Any = Field(default_factory=outer_factory)
    buffer_transforms: Any = Field(default_factory=_default_dict_transforms)
    edge_requests: Any = Field(default_factory=_default_edge_requests)
    vertex_observations: list[VertexObservation] = Field(
        default_factory=_default_vertex_observations
    )
    transform_observations: list[TransformObservation] = Field(
        default_factory=_default_transform_observations
    )
    edge_intents: list[EdgeIntent] = Field(default_factory=_default_edge_intents)

    def record_vertex_observation(
        self, *, vertex_name: str, location: LocationIndex, vertex: dict, ctx: dict
    ) -> None:
        self.vertex_observations.append(
            VertexObservation(
                vertex_name=vertex_name,
                location=location,
                vertex=vertex,
                ctx=ctx,
                provenance=ProvenancePath.from_lindex(location),
            )
        )

    def record_transform_observation(
        self, *, location: LocationIndex, payload: TransformPayload
    ) -> None:
        self.transform_observations.append(
            TransformObservation(
                location=location,
                payload=payload,
                provenance=ProvenancePath.from_lindex(location),
            )
        )

    def record_edge_intent(self, *, edge: Any, location: LocationIndex) -> None:
        self.edge_intents.append(
            EdgeIntent(
                edge=edge,
                location=location,
                provenance=ProvenancePath.from_lindex(location),
            )
        )

GraphAssemblyResult

Bases: ConfigBaseModel

Result of graph assembly phase.

Source code in graflo/architecture/graph_types.py

class GraphAssemblyResult(ConfigBaseModel):
    """Result of graph assembly phase."""

    entities: Any = Field(default_factory=dd_factory)

GraphContainer

Bases: ConfigBaseModel

Container for graph data including vertices and edges.

Attributes:

Name	Type	Description
vertices	dict[str, list]	Dictionary mapping vertex names to lists of vertex data
edges	dict[tuple[str, str, str | None], list]	Dictionary mapping edge IDs to lists of edge data
linear	list[defaultdict[str | tuple[str, str, str | None], list[Any]]]	List of default dictionaries containing linear data

Source code in graflo/architecture/graph_types.py

class GraphContainer(ConfigBaseModel):
    """Container for graph data including vertices and edges.

    Attributes:
        vertices: Dictionary mapping vertex names to lists of vertex data
        edges: Dictionary mapping edge IDs to lists of edge data
        linear: List of default dictionaries containing linear data
    """

    vertices: dict[str, list] = Field(default_factory=dict)
    edges: dict[tuple[str, str, str | None], list] = Field(default_factory=dict)
    linear: list[defaultdict[str | tuple[str, str, str | None], list[Any]]] = Field(
        default_factory=list
    )

    def items(self):
        """Get an ItemsView of the container's contents."""
        return ItemsView(self)

    def pick_unique(self):
        """Remove duplicate entries from vertices and edges."""
        for k, v in self.vertices.items():
            self.vertices[k] = _pick_unique_dict(v)
        for k, v in self.edges.items():
            self.edges[k] = _pick_unique_dict(v)

    def loop_over_relations(self, edge_def: tuple[str, str, str | None]):
        """Iterate over edges matching the given edge definition.

        Args:
            edge_def: Tuple of (source, target, optional_relation)

        Returns:
            Generator yielding matching edge IDs
        """
        source, target, _ = edge_def
        return (ed for ed in self.edges if source == ed[0] and target == ed[1])

    @classmethod
    def from_docs_list(
        cls, list_default_dicts: list[defaultdict[GraphEntity, list]]
    ) -> GraphContainer:
        """Create a GraphContainer from a list of default dictionaries.

        Args:
            list_default_dicts: List of default dictionaries containing vertex and edge data

        Returns:
            New GraphContainer instance

        Raises:
            AssertionError: If edge IDs are not properly formatted
        """
        vdict: defaultdict[str, list] = defaultdict(list)
        edict: defaultdict[tuple[str, str, str | None], list] = defaultdict(list)

        for d in list_default_dicts:
            for k, v in d.items():
                if isinstance(k, str):
                    vdict[k].extend(v)
                elif isinstance(k, tuple):
                    if not (
                        len(k) == 3
                        and all(isinstance(item, str) for item in k[:-1])
                        and isinstance(k[-1], (str, type(None)))
                    ):
                        raise ValueError(
                            f"edge key must be (str, str, str|None), got {k}"
                        )
                    edict[k].extend(v)
        return GraphContainer(
            vertices=dict(vdict.items()),
            edges=dict(edict.items()),
            linear=list_default_dicts,
        )

from_docs_list(list_default_dicts) classmethod

Create a GraphContainer from a list of default dictionaries.

Parameters:

Name	Type	Description	Default
list_default_dicts	list[defaultdict[GraphEntity, list]]	List of default dictionaries containing vertex and edge data	required

Returns:

Type	Description
GraphContainer	New GraphContainer instance

Raises:

Type	Description
AssertionError	If edge IDs are not properly formatted

Source code in graflo/architecture/graph_types.py

@classmethod
def from_docs_list(
    cls, list_default_dicts: list[defaultdict[GraphEntity, list]]
) -> GraphContainer:
    """Create a GraphContainer from a list of default dictionaries.

    Args:
        list_default_dicts: List of default dictionaries containing vertex and edge data

    Returns:
        New GraphContainer instance

    Raises:
        AssertionError: If edge IDs are not properly formatted
    """
    vdict: defaultdict[str, list] = defaultdict(list)
    edict: defaultdict[tuple[str, str, str | None], list] = defaultdict(list)

    for d in list_default_dicts:
        for k, v in d.items():
            if isinstance(k, str):
                vdict[k].extend(v)
            elif isinstance(k, tuple):
                if not (
                    len(k) == 3
                    and all(isinstance(item, str) for item in k[:-1])
                    and isinstance(k[-1], (str, type(None)))
                ):
                    raise ValueError(
                        f"edge key must be (str, str, str|None), got {k}"
                    )
                edict[k].extend(v)
    return GraphContainer(
        vertices=dict(vdict.items()),
        edges=dict(edict.items()),
        linear=list_default_dicts,
    )

items()

Get an ItemsView of the container's contents.

Source code in graflo/architecture/graph_types.py

def items(self):
    """Get an ItemsView of the container's contents."""
    return ItemsView(self)

loop_over_relations(edge_def)

Iterate over edges matching the given edge definition.

Parameters:

Name	Type	Description	Default
edge_def	tuple[str, str, str | None]	Tuple of (source, target, optional_relation)	required

Returns:

Type	Description
	Generator yielding matching edge IDs

Source code in graflo/architecture/graph_types.py

def loop_over_relations(self, edge_def: tuple[str, str, str | None]):
    """Iterate over edges matching the given edge definition.

    Args:
        edge_def: Tuple of (source, target, optional_relation)

    Returns:
        Generator yielding matching edge IDs
    """
    source, target, _ = edge_def
    return (ed for ed in self.edges if source == ed[0] and target == ed[1])

pick_unique()

Remove duplicate entries from vertices and edges.

Source code in graflo/architecture/graph_types.py

def pick_unique(self):
    """Remove duplicate entries from vertices and edges."""
    for k, v in self.vertices.items():
        self.vertices[k] = _pick_unique_dict(v)
    for k, v in self.edges.items():
        self.edges[k] = _pick_unique_dict(v)

Index

Bases: ConfigBaseModel

Configuration for database indexes.

Attributes:

Name	Type	Description
name	str | None	Optional name of the index
fields	list[str]	List of fields to index
unique	bool	Whether the index enforces uniqueness
type	IndexType	Type of index to create
deduplicate	bool	Whether to deduplicate index entries
sparse	bool	Whether to create a sparse index
exclude_edge_endpoints	bool	Whether to exclude edge endpoints from index

Source code in graflo/architecture/graph_types.py

class Index(ConfigBaseModel):
    """Configuration for database indexes.

    Attributes:
        name: Optional name of the index
        fields: List of fields to index
        unique: Whether the index enforces uniqueness
        type: Type of index to create
        deduplicate: Whether to deduplicate index entries
        sparse: Whether to create a sparse index
        exclude_edge_endpoints: Whether to exclude edge endpoints from index
    """

    name: str | None = Field(
        default=None,
        description="Optional index name. For edges, can reference a vertex name for composite fields.",
    )
    fields: list[str] = Field(
        default_factory=list,
        description="List of field names included in this index.",
    )
    unique: bool = Field(
        default=True,
        description="If True, index enforces uniqueness on the field combination.",
    )
    type: IndexType = Field(
        default=IndexType.PERSISTENT,
        description="Index type (PERSISTENT, HASH, SKIPLIST, FULLTEXT).",
    )
    deduplicate: bool = Field(
        default=True,
        description="Whether to deduplicate index entries (e.g. ArangoDB).",
    )
    sparse: bool = Field(
        default=False,
        description="If True, create a sparse index (exclude null/missing values).",
    )
    exclude_edge_endpoints: bool = Field(
        default=False,
        description="If True, do not add _from/_to to edge index (e.g. ArangoDB).",
    )

    def __iter__(self):
        """Iterate over the indexed fields."""
        return iter(self.fields)

    def db_form(self, db_type: DBType) -> dict:
        """Convert index configuration to database-specific format.

        Args:
            db_type: Type of database (ARANGO or NEO4J)

        Returns:
            Dictionary of index configuration in database-specific format

        Raises:
            ValueError: If db_type is not supported
        """
        r = dict(self.to_dict())
        if db_type == DBType.ARANGO:
            r.pop("name", None)
            r.pop("exclude_edge_endpoints", None)
        return r

__iter__()

Iterate over the indexed fields.

Source code in graflo/architecture/graph_types.py

def __iter__(self):
    """Iterate over the indexed fields."""
    return iter(self.fields)

db_form(db_type)

Convert index configuration to database-specific format.

Parameters:

Name	Type	Description	Default
db_type	DBType	Type of database (ARANGO or NEO4J)	required

Returns:

Type	Description
dict	Dictionary of index configuration in database-specific format

Raises:

Type	Description
ValueError	If db_type is not supported

Source code in graflo/architecture/graph_types.py

def db_form(self, db_type: DBType) -> dict:
    """Convert index configuration to database-specific format.

    Args:
        db_type: Type of database (ARANGO or NEO4J)

    Returns:
        Dictionary of index configuration in database-specific format

    Raises:
        ValueError: If db_type is not supported
    """
    r = dict(self.to_dict())
    if db_type == DBType.ARANGO:
        r.pop("name", None)
        r.pop("exclude_edge_endpoints", None)
    return r

IndexType

Bases: BaseEnum

Types of database indexes supported.

PERSISTENT: Standard persistent index HASH: Hash-based index for fast lookups SKIPLIST: Sorted index using skip list data structure FULLTEXT: Index optimized for text search

Source code in graflo/architecture/graph_types.py

class IndexType(BaseEnum):
    """Types of database indexes supported.

    PERSISTENT: Standard persistent index
    HASH: Hash-based index for fast lookups
    SKIPLIST: Sorted index using skip list data structure
    FULLTEXT: Index optimized for text search
    """

    PERSISTENT = "persistent"
    HASH = "hash"
    SKIPLIST = "skiplist"
    FULLTEXT = "fulltext"

ItemsView

View class for iterating over vertices and edges in a GraphContainer.

Source code in graflo/architecture/graph_types.py

class ItemsView:
    """View class for iterating over vertices and edges in a GraphContainer."""

    def __init__(self, gc: GraphContainer):
        self._dictlike = gc

    def __iter__(self):
        """Iterate over vertices and edges in the container."""
        for key in self._dictlike.vertices:
            yield key, self._dictlike.vertices[key]
        for key in self._dictlike.edges:
            yield key, self._dictlike.edges[key]

__iter__()

Iterate over vertices and edges in the container.

Source code in graflo/architecture/graph_types.py

def __iter__(self):
    """Iterate over vertices and edges in the container."""
    for key in self._dictlike.vertices:
        yield key, self._dictlike.vertices[key]
    for key in self._dictlike.edges:
        yield key, self._dictlike.edges[key]

LocationIndex

Bases: ConfigBaseModel

Immutable location index for nested graph traversal.

Source code in graflo/architecture/graph_types.py

class LocationIndex(ConfigBaseModel):
    """Immutable location index for nested graph traversal."""

    model_config = ConfigDict(frozen=True)

    path: tuple[str | int | None, ...] = Field(default_factory=tuple)

    @model_validator(mode="before")
    @classmethod
    def accept_tuple(cls, data: Any) -> Any:
        """Accept a single tuple as positional path (e.g. LocationIndex((0,)))."""
        if isinstance(data, tuple):
            return {"path": data}
        return data

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Allow LocationIndex((0,)) or LocationIndex(path=(0,))."""
        if args and len(args) == 1 and isinstance(args[0], tuple) and not kwargs:
            kwargs = {"path": args[0]}
        super().__init__(**kwargs)

    def extend(self, extension: tuple[str | int | None, ...]) -> LocationIndex:
        return LocationIndex(path=(*self.path, *extension))

    def depth(self) -> int:
        return len(self.path)

    def congruence_measure(self, other: LocationIndex) -> int:
        neq_position = 0
        for step_a, step_b in zip(self.path, other.path):
            if step_a != step_b:
                break
            neq_position += 1
        return neq_position

    def filter(self, lindex_list: list[LocationIndex]) -> list[LocationIndex]:
        return [
            t
            for t in lindex_list
            if t.depth() >= self.depth() and t.path[: self.depth()] == self.path
        ]

    def __lt__(self, other: LocationIndex) -> bool:
        return len(self.path) < len(other.path)

    def __contains__(self, item: object) -> bool:
        return item in self.path

    def __len__(self) -> int:
        return len(self.path)

    def __iter__(self):
        return iter(self.path)

    def __getitem__(self, item: int | slice):
        return self.path[item]

__init__(*args, **kwargs)

Allow LocationIndex((0,)) or LocationIndex(path=(0,)).

Source code in graflo/architecture/graph_types.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Allow LocationIndex((0,)) or LocationIndex(path=(0,))."""
    if args and len(args) == 1 and isinstance(args[0], tuple) and not kwargs:
        kwargs = {"path": args[0]}
    super().__init__(**kwargs)

accept_tuple(data) classmethod

Accept a single tuple as positional path (e.g. LocationIndex((0,))).

Source code in graflo/architecture/graph_types.py

@model_validator(mode="before")
@classmethod
def accept_tuple(cls, data: Any) -> Any:
    """Accept a single tuple as positional path (e.g. LocationIndex((0,)))."""
    if isinstance(data, tuple):
        return {"path": data}
    return data

ProvenancePath

Bases: ConfigBaseModel

Explicit provenance path for extracted observations.

Source code in graflo/architecture/graph_types.py

class ProvenancePath(ConfigBaseModel):
    """Explicit provenance path for extracted observations."""

    path: tuple[str | int | None, ...] = Field(default_factory=tuple)

    @classmethod
    def from_lindex(cls, lindex: LocationIndex) -> ProvenancePath:
        return cls(path=lindex.path)

TransformObservation

Bases: ConfigBaseModel

Typed transform observation emitted during extraction.

Source code in graflo/architecture/graph_types.py

class TransformObservation(ConfigBaseModel):
    """Typed transform observation emitted during extraction."""

    location: LocationIndex
    payload: TransformPayload
    provenance: ProvenancePath

TransformPayload

Bases: ConfigBaseModel

Typed transform output shared between extraction and assembly phases.

Source code in graflo/architecture/graph_types.py

class TransformPayload(ConfigBaseModel):
    """Typed transform output shared between extraction and assembly phases."""

    named: dict[str, Any] = Field(default_factory=dict)
    positional: tuple[Any, ...] = Field(default_factory=tuple)

    @classmethod
    def from_result(cls, result: Any) -> TransformPayload:
        if isinstance(result, dict):
            return cls(named=dict(result))
        if isinstance(result, tuple):
            return cls(positional=tuple(result))
        return cls(positional=(result,))

    def context_doc(self) -> dict[str, Any]:
        """Named values available for weight and relation extraction."""
        return dict(self.named)

context_doc()

Named values available for weight and relation extraction.

Source code in graflo/architecture/graph_types.py

def context_doc(self) -> dict[str, Any]:
    """Named values available for weight and relation extraction."""
    return dict(self.named)

VertexObservation

Bases: ConfigBaseModel

Typed vertex observation emitted during extraction.

Source code in graflo/architecture/graph_types.py

class VertexObservation(ConfigBaseModel):
    """Typed vertex observation emitted during extraction."""

    vertex_name: str
    location: LocationIndex
    vertex: dict[str, Any]
    ctx: dict[str, Any]
    provenance: ProvenancePath

VertexRep

Bases: ConfigBaseModel

Context for graph transformation actions.

Attributes:

Name	Type	Description
vertex	dict	doc representing a vertex
ctx	dict	context (for edge definition upstream

Source code in graflo/architecture/graph_types.py

class VertexRep(ConfigBaseModel):
    """Context for graph transformation actions.

    Attributes:
        vertex: doc representing a vertex
        ctx: context (for edge definition upstream
    """

    model_config = ConfigDict(kw_only=True)  # type: ignore[assignment]

    vertex: dict
    ctx: dict

Weight

Bases: ABCFields

Defines weight configuration for edges.

Attributes:

Name	Type	Description
map	dict	Dictionary mapping field values to weights
filter	dict	Dictionary of filter conditions for weights

Source code in graflo/architecture/graph_types.py

class Weight(ABCFields):
    """Defines weight configuration for edges.

    Attributes:
        map: Dictionary mapping field values to weights
        filter: Dictionary of filter conditions for weights
    """

    map: dict = Field(
        default_factory=dict,
        description="Mapping of field values to weight values for vertex-based edge attributes.",
    )
    filter: dict = Field(
        default_factory=dict,
        description="Filter conditions applied when resolving vertex-based weights.",
    )

dd_factory()

Create a default dictionary for graph entity data.

Source code in graflo/architecture/graph_types.py

def dd_factory() -> defaultdict[GraphEntity, list]:
    """Create a default dictionary for graph entity data."""
    return defaultdict(list)

inner_factory_vertex()

Create a default dictionary for vertex data.

Source code in graflo/architecture/graph_types.py

def inner_factory_vertex() -> defaultdict[LocationIndex, list]:
    """Create a default dictionary for vertex data."""
    return defaultdict(list)

outer_factory()

Create a nested default dictionary for vertex data.

Source code in graflo/architecture/graph_types.py

def outer_factory() -> defaultdict[str, defaultdict[LocationIndex, list]]:
    """Create a nested default dictionary for vertex data."""
    return defaultdict(inner_factory_vertex)