graflo.architecture

Graph database architecture components.

This package defines the core architecture components for graph database operations, including schema management, resource handling, and data transformations.

Key Components

• Schema: Graph database schema definition and management
• Resource: Data resource management and processing
• Transform: Data transformation and standardization
• Vertex: Vertex collection configuration
• Edge: Edge collection configuration

Example

from graflo.architecture import Schema, Resource schema = Schema( ... general={"name": "my_graph", "version": "1.0"}, ... vertex_config=vertex_config, ... edge_config=edge_config ... ) resource = Resource(name="users", data=user_data)

Edge dataclass

Bases: BaseDataclass

Represents an edge in the graph database.

An edge connects two vertices and can have various configurations for indexing, weights, and relationship types.

Attributes:

Name	Type	Description
source	str	Source vertex name
target	str	Target vertex name
indexes	list[Index]	List of indexes for the edge
weights	Union[WeightConfig, None]	Optional weight configuration
relation	str | None	Optional relation name (for Neo4j)
purpose	str | None	Optional purpose for utility collections
match_source	str | None	Optional source discriminant field
match_target	str | None	Optional target discriminant field
type	EdgeType	Edge type (DIRECT or INDIRECT)
aux	bool	Whether this is an auxiliary edge
by	str | None	Optional vertex name for indirect edges
graph_name	str | None	Optional graph name (ArangoDB only, set in finish_init)
database_name	str | None	Optional database-specific edge identifier (ArangoDB only, set in finish_init). For ArangoDB, this corresponds to the edge collection name.

Source code in graflo/architecture/edge.py

@dataclasses.dataclass
class Edge(BaseDataclass):
    """Represents an edge in the graph database.

    An edge connects two vertices and can have various configurations for
    indexing, weights, and relationship types.

    Attributes:
        source: Source vertex name
        target: Target vertex name
        indexes: List of indexes for the edge
        weights: Optional weight configuration
        relation: Optional relation name (for Neo4j)
        purpose: Optional purpose for utility collections
        match_source: Optional source discriminant field
        match_target: Optional target discriminant field
        type: Edge type (DIRECT or INDIRECT)
        aux: Whether this is an auxiliary edge
        by: Optional vertex name for indirect edges
        graph_name: Optional graph name (ArangoDB only, set in finish_init)
        database_name: Optional database-specific edge identifier (ArangoDB only, set in finish_init).
                       For ArangoDB, this corresponds to the edge collection name.
    """

    source: str
    target: str
    indexes: list[Index] = dataclasses.field(default_factory=list)
    weights: Union[WeightConfig, None] = (
        None  # Using Union for dataclass_wizard compatibility
    )

    # relation represents Class in neo4j, for arango it becomes a weight
    relation: str | None = None
    # field that contains Class or relation
    relation_field: str | None = None
    relation_from_key: bool = False

    # used to create extra utility collections between the same type of vertices (A, B)
    purpose: str | None = None

    match_source: str | None = None
    match_target: str | None = None
    exclude_source: str | None = None
    exclude_target: str | None = None
    match: str | None = None

    type: EdgeType = EdgeType.DIRECT

    aux: bool = False  # aux=True edges are init in the db but not considered by graflo

    by: str | None = None
    graph_name: str | None = None  # ArangoDB-specific: graph name (set in finish_init)
    database_name: str | None = (
        None  # ArangoDB-specific: edge collection name (set in finish_init)
    )

    def __post_init__(self):
        """Initialize the edge after dataclass initialization."""

        self._source: str | None = None
        self._target: str | None = None

    def finish_init(self, vertex_config: VertexConfig):
        """Complete edge initialization with vertex configuration.

        Sets up edge collections, graph names, and initializes indices based on
        the vertex configuration.

        Args:
            vertex_config: Configuration for vertices

        """
        if self.type == EdgeType.INDIRECT and self.by is not None:
            self.by = vertex_config.vertex_dbname(self.by)

        self._source = vertex_config.vertex_dbname(self.source)
        self._target = vertex_config.vertex_dbname(self.target)

        # ArangoDB-specific: set graph_name and database_name only for ArangoDB
        if vertex_config.db_flavor == DBFlavor.ARANGO:
            graph_name = [
                vertex_config.vertex_dbname(self.source),
                vertex_config.vertex_dbname(self.target),
            ]
            if self.purpose is not None:
                graph_name += [self.purpose]
            self.graph_name = "_".join(graph_name + ["graph"])
            self.database_name = "_".join(graph_name + ["edges"])

        # TigerGraph requires named edge types (relations), so assign default if missing
        if vertex_config.db_flavor == DBFlavor.TIGERGRAPH and self.relation is None:
            # Use default relation name for TigerGraph
            # TigerGraph requires all edges to have a named type (relation)
            self.relation = DEFAULT_TIGERGRAPH_RELATION

        self._init_indices(vertex_config)

    def _init_indices(self, vc: VertexConfig):
        """Initialize indices for the edge.

        Args:
            vc: Vertex configuration
        """
        self.indexes = [self._init_index(index, vc) for index in self.indexes]

    def _init_index(self, index: Index, vc: VertexConfig) -> Index:
        """Initialize a single index for the edge.

        Args:
            index: Index to initialize
            vc: Vertex configuration

        Returns:
            Index: Initialized index

        Note:
            Default behavior for edge indices: adds ["_from", "_to"] for uniqueness
            in ArangoDB.
        """
        index_fields = []

        # "@" is reserved : quick hack - do not reinit the index twice
        if any("@" in f for f in index.fields):
            return index
        if index.name is None:
            index_fields += index.fields
        else:
            # add index over a vertex of index.name
            if index.fields:
                fields = index.fields
            else:
                fields = vc.index(index.name).fields
            index_fields += [f"{index.name}@{x}" for x in fields]

        if not index.exclude_edge_endpoints and vc.db_flavor == DBFlavor.ARANGO:
            if all([item not in index_fields for item in ["_from", "_to"]]):
                index_fields = ["_from", "_to"] + index_fields

        index.fields = index_fields
        return index

    @property
    def edge_name_dyad(self):
        """Get the edge name as a dyad (source, target).

        Returns:
            tuple[str, str]: Source and target vertex names
        """
        return self.source, self.target

    @property
    def edge_id(self) -> EdgeId:
        """Get the edge ID.

        Returns:
            EdgeId: Tuple of (source, target, purpose)
        """
        return self.source, self.target, self.purpose

edge_id property

Get the edge ID.

Returns:

Name	Type	Description
EdgeId	EdgeId	Tuple of (source, target, purpose)

edge_name_dyad property

Get the edge name as a dyad (source, target).

Returns:

Type	Description
	tuple[str, str]: Source and target vertex names

__post_init__()

Initialize the edge after dataclass initialization.

Source code in graflo/architecture/edge.py

def __post_init__(self):
    """Initialize the edge after dataclass initialization."""

    self._source: str | None = None
    self._target: str | None = None

finish_init(vertex_config)

Complete edge initialization with vertex configuration.

Sets up edge collections, graph names, and initializes indices based on the vertex configuration.

Parameters:

Name	Type	Description	Default
vertex_config	VertexConfig	Configuration for vertices	required

Source code in graflo/architecture/edge.py

def finish_init(self, vertex_config: VertexConfig):
    """Complete edge initialization with vertex configuration.

    Sets up edge collections, graph names, and initializes indices based on
    the vertex configuration.

    Args:
        vertex_config: Configuration for vertices

    """
    if self.type == EdgeType.INDIRECT and self.by is not None:
        self.by = vertex_config.vertex_dbname(self.by)

    self._source = vertex_config.vertex_dbname(self.source)
    self._target = vertex_config.vertex_dbname(self.target)

    # ArangoDB-specific: set graph_name and database_name only for ArangoDB
    if vertex_config.db_flavor == DBFlavor.ARANGO:
        graph_name = [
            vertex_config.vertex_dbname(self.source),
            vertex_config.vertex_dbname(self.target),
        ]
        if self.purpose is not None:
            graph_name += [self.purpose]
        self.graph_name = "_".join(graph_name + ["graph"])
        self.database_name = "_".join(graph_name + ["edges"])

    # TigerGraph requires named edge types (relations), so assign default if missing
    if vertex_config.db_flavor == DBFlavor.TIGERGRAPH and self.relation is None:
        # Use default relation name for TigerGraph
        # TigerGraph requires all edges to have a named type (relation)
        self.relation = DEFAULT_TIGERGRAPH_RELATION

    self._init_indices(vertex_config)

EdgeConfig dataclass

Bases: BaseDataclass

Configuration for managing collections of edges.

This class manages a collection of edges, providing methods for accessing and manipulating edge configurations.

Attributes:

Name	Type	Description
edges	list[Edge]	List of edge configurations

Source code in graflo/architecture/edge.py

@dataclasses.dataclass
class EdgeConfig(BaseDataclass):
    """Configuration for managing collections of edges.

    This class manages a collection of edges, providing methods for accessing
    and manipulating edge configurations.

    Attributes:
        edges: List of edge configurations
    """

    edges: list[Edge] = dataclasses.field(default_factory=list)

    def __post_init__(self):
        """Initialize the edge configuration.

        Creates internal mapping of edge IDs to edge configurations.
        """
        self._edges_map: dict[EdgeId, Edge] = {e.edge_id: e for e in self.edges}

    def finish_init(self, vc: VertexConfig):
        """Complete initialization of all edges with vertex configuration.

        Args:
            vc: Vertex configuration
        """
        for e in self.edges:
            e.finish_init(vc)

    def edges_list(self, include_aux=False):
        """Get list of edges.

        Args:
            include_aux: Whether to include auxiliary edges

        Returns:
            generator: Generator yielding edge configurations
        """
        return (e for e in self._edges_map.values() if include_aux or not e.aux)

    def edges_items(self, include_aux=False):
        """Get items of edges.

        Args:
            include_aux: Whether to include auxiliary edges

        Returns:
            generator: Generator yielding (edge_id, edge) tuples
        """
        return (
            (eid, e) for eid, e in self._edges_map.items() if include_aux or not e.aux
        )

    def __contains__(self, item: EdgeId | Edge):
        """Check if edge exists in configuration.

        Args:
            item: Edge ID or Edge instance to check

        Returns:
            bool: True if edge exists, False otherwise
        """
        if isinstance(item, Edge):
            eid = item.edge_id
        else:
            eid = item

        if eid in self._edges_map:
            return True
        else:
            return False

    def update_edges(self, edge: Edge, vertex_config: VertexConfig):
        """Update edge configuration.

        Args:
            edge: Edge configuration to update
            vertex_config: Vertex configuration
        """
        if edge.edge_id in self._edges_map:
            self._edges_map[edge.edge_id].update(edge)
        else:
            self._edges_map[edge.edge_id] = edge
        self._edges_map[edge.edge_id].finish_init(vertex_config=vertex_config)

    @property
    def vertices(self):
        """Get set of vertex names involved in edges.

        Returns:
            set[str]: Set of vertex names
        """
        return {e.source for e in self.edges} | {e.target for e in self.edges}

vertices property

Get set of vertex names involved in edges.

Returns:

Type	Description
	set[str]: Set of vertex names

__contains__(item)

Check if edge exists in configuration.

Parameters:

Name	Type	Description	Default
item	EdgeId | Edge	Edge ID or Edge instance to check	required

Returns:

Name	Type	Description
bool		True if edge exists, False otherwise

Source code in graflo/architecture/edge.py

def __contains__(self, item: EdgeId | Edge):
    """Check if edge exists in configuration.

    Args:
        item: Edge ID or Edge instance to check

    Returns:
        bool: True if edge exists, False otherwise
    """
    if isinstance(item, Edge):
        eid = item.edge_id
    else:
        eid = item

    if eid in self._edges_map:
        return True
    else:
        return False

__post_init__()

Initialize the edge configuration.

Creates internal mapping of edge IDs to edge configurations.

Source code in graflo/architecture/edge.py

def __post_init__(self):
    """Initialize the edge configuration.

    Creates internal mapping of edge IDs to edge configurations.
    """
    self._edges_map: dict[EdgeId, Edge] = {e.edge_id: e for e in self.edges}

edges_items(include_aux=False)

Get items of edges.

Parameters:

Name	Type	Description	Default
include_aux		Whether to include auxiliary edges	False

Returns:

Name	Type	Description
generator		Generator yielding (edge_id, edge) tuples

Source code in graflo/architecture/edge.py

def edges_items(self, include_aux=False):
    """Get items of edges.

    Args:
        include_aux: Whether to include auxiliary edges

    Returns:
        generator: Generator yielding (edge_id, edge) tuples
    """
    return (
        (eid, e) for eid, e in self._edges_map.items() if include_aux or not e.aux
    )

edges_list(include_aux=False)

Get list of edges.

Parameters:

Name	Type	Description	Default
include_aux		Whether to include auxiliary edges	False

Returns:

Name	Type	Description
generator		Generator yielding edge configurations

Source code in graflo/architecture/edge.py

def edges_list(self, include_aux=False):
    """Get list of edges.

    Args:
        include_aux: Whether to include auxiliary edges

    Returns:
        generator: Generator yielding edge configurations
    """
    return (e for e in self._edges_map.values() if include_aux or not e.aux)

finish_init(vc)

Complete initialization of all edges with vertex configuration.

Parameters:

Name	Type	Description	Default
vc	VertexConfig	Vertex configuration	required

Source code in graflo/architecture/edge.py

def finish_init(self, vc: VertexConfig):
    """Complete initialization of all edges with vertex configuration.

    Args:
        vc: Vertex configuration
    """
    for e in self.edges:
        e.finish_init(vc)

update_edges(edge, vertex_config)

Update edge configuration.

Parameters:

Name	Type	Description	Default
edge	Edge	Edge configuration to update	required
vertex_config	VertexConfig	Vertex configuration	required

Source code in graflo/architecture/edge.py

def update_edges(self, edge: Edge, vertex_config: VertexConfig):
    """Update edge configuration.

    Args:
        edge: Edge configuration to update
        vertex_config: Vertex configuration
    """
    if edge.edge_id in self._edges_map:
        self._edges_map[edge.edge_id].update(edge)
    else:
        self._edges_map[edge.edge_id] = edge
    self._edges_map[edge.edge_id].finish_init(vertex_config=vertex_config)

FieldType

Bases: BaseEnum

Supported field types for graph databases.

These types are primarily used for TigerGraph, which requires explicit field types. Other databases (ArangoDB, Neo4j) may use different type systems or not require types.

Attributes:

Name	Type	Description
INT		Integer type
UINT		Unsigned integer type
FLOAT		Floating point type
DOUBLE		Double precision floating point type
BOOL		Boolean type
STRING		String type
DATETIME		DateTime type

Source code in graflo/architecture/vertex.py

class FieldType(BaseEnum):
    """Supported field types for graph databases.

    These types are primarily used for TigerGraph, which requires explicit field types.
    Other databases (ArangoDB, Neo4j) may use different type systems or not require types.

    Attributes:
        INT: Integer type
        UINT: Unsigned integer type
        FLOAT: Floating point type
        DOUBLE: Double precision floating point type
        BOOL: Boolean type
        STRING: String type
        DATETIME: DateTime type
    """

    INT = "INT"
    UINT = "UINT"
    FLOAT = "FLOAT"
    DOUBLE = "DOUBLE"
    BOOL = "BOOL"
    STRING = "STRING"
    DATETIME = "DATETIME"

Index dataclass

Bases: BaseDataclass

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/onto.py

@dataclasses.dataclass
class Index(BaseDataclass):
    """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 = None
    fields: list[str] = dataclasses.field(default_factory=list)
    unique: bool = True
    type: IndexType = IndexType.PERSISTENT
    deduplicate: bool = True
    sparse: bool = False
    exclude_edge_endpoints: bool = False

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

    def db_form(self, db_type: DBFlavor) -> 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 = self.to_dict()
        if db_type == DBFlavor.ARANGO:
            _ = r.pop("name")
            _ = r.pop("exclude_edge_endpoints")
        elif db_type == DBFlavor.NEO4J:
            pass
        else:
            raise ValueError(f"Unknown db_type {db_type}")

        return r

__iter__()

Iterate over the indexed fields.

Source code in graflo/architecture/onto.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	DBFlavor	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/onto.py

def db_form(self, db_type: DBFlavor) -> 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 = self.to_dict()
    if db_type == DBFlavor.ARANGO:
        _ = r.pop("name")
        _ = r.pop("exclude_edge_endpoints")
    elif db_type == DBFlavor.NEO4J:
        pass
    else:
        raise ValueError(f"Unknown db_type {db_type}")

    return r

Resource dataclass

Bases: BaseDataclass, JSONWizard

Resource configuration and processing.

This class represents a data resource that can be processed and transformed into graph structures. It manages the processing pipeline through actors and handles data encoding, transformation, and mapping.

Attributes:

Name	Type	Description
resource_name	str	Name of the resource
apply	list	List of actors to apply in sequence
encoding	EncodingType	Data encoding type (default: UTF_8)
merge_collections	list[str]	List of collections to merge
extra_weights	list[Edge]	List of additional edge weights
types	dict[str, str]	Dictionary of field type mappings
root	dict[str, str]	Root actor wrapper for processing
vertex_config	dict[str, str]	Configuration for vertices
edge_config	dict[str, str]	Configuration for edges

Source code in graflo/architecture/resource.py

@dataclasses.dataclass(kw_only=True)
class Resource(BaseDataclass, JSONWizard):
    """Resource configuration and processing.

    This class represents a data resource that can be processed and transformed
    into graph structures. It manages the processing pipeline through actors
    and handles data encoding, transformation, and mapping.

    Attributes:
        resource_name: Name of the resource
        apply: List of actors to apply in sequence
        encoding: Data encoding type (default: UTF_8)
        merge_collections: List of collections to merge
        extra_weights: List of additional edge weights
        types: Dictionary of field type mappings
        root: Root actor wrapper for processing
        vertex_config: Configuration for vertices
        edge_config: Configuration for edges
    """

    resource_name: str
    apply: list
    encoding: EncodingType = EncodingType.UTF_8
    merge_collections: list[str] = dataclasses.field(default_factory=list)
    extra_weights: list[Edge] = dataclasses.field(default_factory=list)
    types: dict[str, str] = dataclasses.field(default_factory=dict)
    edge_greedy: bool = True

    def __post_init__(self):
        """Initialize the resource after dataclass initialization.

        Sets up the actor wrapper and type mappings. Evaluates type expressions
        for field type casting.

        Raises:
            Exception: If type evaluation fails for any field
        """
        self.root = ActorWrapper(*self.apply)
        self._types: dict[str, Callable] = dict()
        self.vertex_config: VertexConfig
        self.edge_config: EdgeConfig
        for k, v in self.types.items():
            try:
                self._types[k] = eval(v)
            except Exception as ex:
                logger.error(
                    f"For resource {self.name} for field {k} failed to cast type {v} : {ex}"
                )

    @property
    def name(self):
        """Get the resource name.

        Returns:
            str: Name of the resource
        """
        return self.resource_name

    def finish_init(
        self,
        vertex_config: VertexConfig,
        edge_config: EdgeConfig,
        transforms: dict[str, ProtoTransform],
    ):
        """Complete resource initialization.

        Initializes the resource with vertex and edge configurations,
        and sets up the processing pipeline.

        Args:
            vertex_config: Configuration for vertices
            edge_config: Configuration for edges
            transforms: Dictionary of available transforms
        """
        self.vertex_config = vertex_config
        self.edge_config = edge_config

        logger.debug(f"total resource actor count : {self.root.count()}")
        self.root.finish_init(
            vertex_config=vertex_config,
            transforms=transforms,
            edge_config=edge_config,
            edge_greedy=self.edge_greedy,
        )

        logger.debug(f"total resource actor count (after 2 finit): {self.root.count()}")

        for e in self.extra_weights:
            e.finish_init(vertex_config)

    def __call__(self, doc: dict) -> defaultdict[GraphEntity, list]:
        """Process a document through the resource pipeline.

        Args:
            doc: Document to process

        Returns:
            defaultdict[GraphEntity, list]: Processed graph entities
        """
        ctx = ActionContext()
        ctx = self.root(ctx, doc=doc)
        acc = self.root.normalize_ctx(ctx)
        return acc

    def count(self):
        """Get the total number of actors in the resource.

        Returns:
            int: Number of actors
        """
        return self.root.count()

name property

Get the resource name.

Returns:

Name	Type	Description
str		Name of the resource

__call__(doc)

Process a document through the resource pipeline.

Parameters:

Name	Type	Description	Default
doc	dict	Document to process	required

Returns:

Type	Description
defaultdict[GraphEntity, list]	defaultdict[GraphEntity, list]: Processed graph entities

Source code in graflo/architecture/resource.py

def __call__(self, doc: dict) -> defaultdict[GraphEntity, list]:
    """Process a document through the resource pipeline.

    Args:
        doc: Document to process

    Returns:
        defaultdict[GraphEntity, list]: Processed graph entities
    """
    ctx = ActionContext()
    ctx = self.root(ctx, doc=doc)
    acc = self.root.normalize_ctx(ctx)
    return acc

__post_init__()

Initialize the resource after dataclass initialization.

Sets up the actor wrapper and type mappings. Evaluates type expressions for field type casting.

Raises:

Type	Description
Exception	If type evaluation fails for any field

Source code in graflo/architecture/resource.py

def __post_init__(self):
    """Initialize the resource after dataclass initialization.

    Sets up the actor wrapper and type mappings. Evaluates type expressions
    for field type casting.

    Raises:
        Exception: If type evaluation fails for any field
    """
    self.root = ActorWrapper(*self.apply)
    self._types: dict[str, Callable] = dict()
    self.vertex_config: VertexConfig
    self.edge_config: EdgeConfig
    for k, v in self.types.items():
        try:
            self._types[k] = eval(v)
        except Exception as ex:
            logger.error(
                f"For resource {self.name} for field {k} failed to cast type {v} : {ex}"
            )

count()

Get the total number of actors in the resource.

Returns:

Name	Type	Description
int		Number of actors

Source code in graflo/architecture/resource.py

def count(self):
    """Get the total number of actors in the resource.

    Returns:
        int: Number of actors
    """
    return self.root.count()

finish_init(vertex_config, edge_config, transforms)

Complete resource initialization.

Initializes the resource with vertex and edge configurations, and sets up the processing pipeline.

Parameters:

Name	Type	Description	Default
vertex_config	VertexConfig	Configuration for vertices	required
edge_config	EdgeConfig	Configuration for edges	required
transforms	dict[str, ProtoTransform]	Dictionary of available transforms	required

Source code in graflo/architecture/resource.py

def finish_init(
    self,
    vertex_config: VertexConfig,
    edge_config: EdgeConfig,
    transforms: dict[str, ProtoTransform],
):
    """Complete resource initialization.

    Initializes the resource with vertex and edge configurations,
    and sets up the processing pipeline.

    Args:
        vertex_config: Configuration for vertices
        edge_config: Configuration for edges
        transforms: Dictionary of available transforms
    """
    self.vertex_config = vertex_config
    self.edge_config = edge_config

    logger.debug(f"total resource actor count : {self.root.count()}")
    self.root.finish_init(
        vertex_config=vertex_config,
        transforms=transforms,
        edge_config=edge_config,
        edge_greedy=self.edge_greedy,
    )

    logger.debug(f"total resource actor count (after 2 finit): {self.root.count()}")

    for e in self.extra_weights:
        e.finish_init(vertex_config)

Schema dataclass

Bases: BaseDataclass

Graph database schema configuration.

This class represents the complete schema configuration for a graph database. It manages resources, vertex configurations, edge configurations, and transforms.

Attributes:

Name	Type	Description
general	SchemaMetadata	Schema metadata and versioning information
vertex_config	VertexConfig	Configuration for vertex collections
edge_config	EdgeConfig	Configuration for edge collections
resources	list[Resource]	List of resource definitions
transforms	dict[str, ProtoTransform]	Dictionary of available transforms
_resources	dict[str, ProtoTransform]	Internal mapping of resource names to resources

Source code in graflo/architecture/schema.py

@dataclasses.dataclass
class Schema(BaseDataclass):
    """Graph database schema configuration.

    This class represents the complete schema configuration for a graph database.
    It manages resources, vertex configurations, edge configurations, and transforms.

    Attributes:
        general: Schema metadata and versioning information
        vertex_config: Configuration for vertex collections
        edge_config: Configuration for edge collections
        resources: List of resource definitions
        transforms: Dictionary of available transforms
        _resources: Internal mapping of resource names to resources
    """

    general: SchemaMetadata
    vertex_config: VertexConfig
    edge_config: EdgeConfig
    resources: list[Resource]
    transforms: dict[str, ProtoTransform] = dataclasses.field(default_factory=dict)

    def __post_init__(self):
        """Initialize the schema after dataclass initialization.

        Sets up transforms, initializes edge configuration, and validates
        resource names for uniqueness.

        Raises:
            ValueError: If duplicate resource names are found
        """
        for name, t in self.transforms.items():
            t.name = name

        self.edge_config.finish_init(self.vertex_config)

        for r in self.resources:
            r.finish_init(
                vertex_config=self.vertex_config,
                edge_config=self.edge_config,
                transforms=self.transforms,
            )

        names = [r.name for r in self.resources]
        c = Counter(names)
        for k, v in c.items():
            if v > 1:
                raise ValueError(f"resource name {k} used {v} times")
        self._resources: dict[str, Resource] = {}
        for r in self.resources:
            self._resources[r.name] = r

    def fetch_resource(self, name: str | None = None) -> Resource:
        """Fetch a resource by name or get the first available resource.

        Args:
            name: Optional name of the resource to fetch

        Returns:
            Resource: The requested resource

        Raises:
            ValueError: If the requested resource is not found or if no resources exist
        """
        _current_resource = None

        if name is not None:
            if name in self._resources:
                _current_resource = self._resources[name]
            else:
                raise ValueError(f"Resource {name} not found")
        else:
            if self._resources:
                _current_resource = self.resources[0]
            else:
                raise ValueError("Empty resource container 😕")
        return _current_resource

__post_init__()

Initialize the schema after dataclass initialization.

Sets up transforms, initializes edge configuration, and validates resource names for uniqueness.

Raises:

Type	Description
ValueError	If duplicate resource names are found

Source code in graflo/architecture/schema.py

def __post_init__(self):
    """Initialize the schema after dataclass initialization.

    Sets up transforms, initializes edge configuration, and validates
    resource names for uniqueness.

    Raises:
        ValueError: If duplicate resource names are found
    """
    for name, t in self.transforms.items():
        t.name = name

    self.edge_config.finish_init(self.vertex_config)

    for r in self.resources:
        r.finish_init(
            vertex_config=self.vertex_config,
            edge_config=self.edge_config,
            transforms=self.transforms,
        )

    names = [r.name for r in self.resources]
    c = Counter(names)
    for k, v in c.items():
        if v > 1:
            raise ValueError(f"resource name {k} used {v} times")
    self._resources: dict[str, Resource] = {}
    for r in self.resources:
        self._resources[r.name] = r

fetch_resource(name=None)

Fetch a resource by name or get the first available resource.

Parameters:

Name	Type	Description	Default
name	str | None	Optional name of the resource to fetch	None

Returns:

Name	Type	Description
Resource	Resource	The requested resource

Raises:

Type	Description
ValueError	If the requested resource is not found or if no resources exist

Source code in graflo/architecture/schema.py

def fetch_resource(self, name: str | None = None) -> Resource:
    """Fetch a resource by name or get the first available resource.

    Args:
        name: Optional name of the resource to fetch

    Returns:
        Resource: The requested resource

    Raises:
        ValueError: If the requested resource is not found or if no resources exist
    """
    _current_resource = None

    if name is not None:
        if name in self._resources:
            _current_resource = self._resources[name]
        else:
            raise ValueError(f"Resource {name} not found")
    else:
        if self._resources:
            _current_resource = self.resources[0]
        else:
            raise ValueError("Empty resource container 😕")
    return _current_resource

Vertex dataclass

Bases: BaseDataclass

Represents a vertex in the graph database.

A vertex is a fundamental unit in the graph that can have fields, indexes, and filters. Fields can be specified as strings, Field objects, or dicts. Internally, fields are stored as Field objects but behave like strings for backward compatibility.

Attributes:

Name	Type	Description
name	str	Name of the vertex
fields	_FieldsType	List of field names (str), Field objects, or dicts. Will be normalized to Field objects internally in post_init. After initialization, this is always list[Field] (type checker sees this).
indexes	list[Index]	List of indexes for the vertex
filters	list[Expression]	List of filter expressions
dbname	str | None	Optional database name (defaults to vertex name)

Examples:

>>> # Backward compatible: list of strings
>>> v1 = Vertex(name="user", fields=["id", "name"])

>>> # Typed fields: list of Field objects
>>> v2 = Vertex(name="user", fields=[
...     Field(name="id", type="INT"),
...     Field(name="name", type="STRING")
... ])

>>> # From dicts (e.g., from YAML/JSON)
>>> v3 = Vertex(name="user", fields=[
...     {"name": "id", "type": "INT"},
...     {"name": "name"}  # defaults to None type
... ])

Source code in graflo/architecture/vertex.py

@dataclasses.dataclass
class Vertex(BaseDataclass):
    """Represents a vertex in the graph database.

    A vertex is a fundamental unit in the graph that can have fields, indexes,
    and filters. Fields can be specified as strings, Field objects, or dicts.
    Internally, fields are stored as Field objects but behave like strings
    for backward compatibility.

    Attributes:
        name: Name of the vertex
        fields: List of field names (str), Field objects, or dicts.
               Will be normalized to Field objects internally in __post_init__.
               After initialization, this is always list[Field] (type checker sees this).
        indexes: List of indexes for the vertex
        filters: List of filter expressions
        dbname: Optional database name (defaults to vertex name)

    Examples:
        >>> # Backward compatible: list of strings
        >>> v1 = Vertex(name="user", fields=["id", "name"])

        >>> # Typed fields: list of Field objects
        >>> v2 = Vertex(name="user", fields=[
        ...     Field(name="id", type="INT"),
        ...     Field(name="name", type="STRING")
        ... ])

        >>> # From dicts (e.g., from YAML/JSON)
        >>> v3 = Vertex(name="user", fields=[
        ...     {"name": "id", "type": "INT"},
        ...     {"name": "name"}  # defaults to None type
        ... ])
    """

    name: str
    fields: _FieldsType = dataclasses.field(default_factory=list)
    indexes: list[Index] = dataclasses.field(default_factory=list)
    filters: list[Expression] = dataclasses.field(default_factory=list)
    dbname: str | None = None

    @staticmethod
    def _parse_string_to_dict(field_str: str) -> dict | None:
        """Parse a string that might be a JSON or Python dict representation.

        Args:
            field_str: String that might be a dict representation

        Returns:
            dict if successfully parsed as dict, None otherwise
        """
        # Try JSON first (handles double-quoted strings)
        try:
            parsed = json.loads(field_str)
            return parsed if isinstance(parsed, dict) else None
        except json.JSONDecodeError:
            pass

        # Try Python literal eval (handles single-quoted strings)
        try:
            parsed = ast.literal_eval(field_str)
            return parsed if isinstance(parsed, dict) else None
        except (ValueError, SyntaxError):
            return None

    @staticmethod
    def _dict_to_field(field_dict: dict) -> Field:
        """Convert a dict to a Field object.

        Args:
            field_dict: Dictionary with 'name' key and optional 'type' key

        Returns:
            Field object

        Raises:
            ValueError: If dict doesn't have 'name' key
        """
        name = field_dict.get("name")
        if name is None:
            raise ValueError(f"Field dict must have 'name' key: {field_dict}")
        return Field(name=name, type=field_dict.get("type"))

    def _normalize_fields(
        self, fields: list[str] | list[Field] | list[dict]
    ) -> list[Field]:
        """Normalize fields to Field objects.

        Converts strings, Field objects, or dicts to Field objects.
        Handles the case where dataclass_wizard may have converted dicts to JSON strings.
        Field objects behave like strings for backward compatibility.

        Args:
            fields: List of strings, Field objects, or dicts

        Returns:
            list[Field]: Normalized list of Field objects (preserving order)
        """
        normalized = []
        for field in fields:
            if isinstance(field, Field):
                normalized.append(field)
            elif isinstance(field, dict):
                normalized.append(self._dict_to_field(field))
            elif isinstance(field, str):
                # Try to parse as dict (JSON or Python literal)
                parsed_dict = self._parse_string_to_dict(field)
                if parsed_dict:
                    normalized.append(self._dict_to_field(parsed_dict))
                else:
                    # Plain field name
                    normalized.append(Field(name=field, type=None))
            else:
                raise TypeError(f"Field must be str, Field, or dict, got {type(field)}")
        return normalized

    @property
    def field_names(self) -> list[str]:
        """Get list of field names (as strings).

        Returns:
            list[str]: List of field names
        """
        return [field.name for field in self.fields]

    def get_fields(self) -> list[Field]:
        return self.fields

    def __post_init__(self):
        """Initialize the vertex after dataclass initialization.

        Sets the database name if not provided, normalizes fields to Field objects,
        and updates fields based on indexes. Field objects behave like strings,
        maintaining backward compatibility.
        """
        if self.dbname is None:
            self.dbname = self.name

        # Normalize fields to Field objects (preserve order)
        self.fields = self._normalize_fields(self.fields)

        # Normalize indexes to Index objects if they're dicts
        normalized_indexes = []
        for idx in self.indexes:
            if isinstance(idx, dict):
                normalized_indexes.append(Index.from_dict(idx))
            else:
                normalized_indexes.append(idx)
        self.indexes = normalized_indexes

        if not self.indexes:
            # Index expects list[str], but Field objects convert to strings automatically
            # via __str__, so we extract names
            self.indexes = [Index(fields=self.field_names)]

        # Collect field names from existing fields (preserve order)
        seen_names = {f.name for f in self.fields}
        # Add index fields that aren't already present (preserve original order, append new)
        for idx in self.indexes:
            for field_name in idx.fields:
                if field_name not in seen_names:
                    # Add new field, preserving order by adding to end
                    self.fields.append(Field(name=field_name, type=None))
                    seen_names.add(field_name)

    def finish_init(self, db_flavor: DBFlavor):
        """Complete initialization of all edges with vertex configuration."""
        self.fields = [
            Field(name=f.name, type=FieldType.STRING)
            if f.type is None and db_flavor == DBFlavor.TIGERGRAPH
            else f
            for f in self.fields
        ]

field_names property

Get list of field names (as strings).

Returns:

Type	Description
list[str]	list[str]: List of field names

__post_init__()

Initialize the vertex after dataclass initialization.

Sets the database name if not provided, normalizes fields to Field objects, and updates fields based on indexes. Field objects behave like strings, maintaining backward compatibility.

Source code in graflo/architecture/vertex.py

def __post_init__(self):
    """Initialize the vertex after dataclass initialization.

    Sets the database name if not provided, normalizes fields to Field objects,
    and updates fields based on indexes. Field objects behave like strings,
    maintaining backward compatibility.
    """
    if self.dbname is None:
        self.dbname = self.name

    # Normalize fields to Field objects (preserve order)
    self.fields = self._normalize_fields(self.fields)

    # Normalize indexes to Index objects if they're dicts
    normalized_indexes = []
    for idx in self.indexes:
        if isinstance(idx, dict):
            normalized_indexes.append(Index.from_dict(idx))
        else:
            normalized_indexes.append(idx)
    self.indexes = normalized_indexes

    if not self.indexes:
        # Index expects list[str], but Field objects convert to strings automatically
        # via __str__, so we extract names
        self.indexes = [Index(fields=self.field_names)]

    # Collect field names from existing fields (preserve order)
    seen_names = {f.name for f in self.fields}
    # Add index fields that aren't already present (preserve original order, append new)
    for idx in self.indexes:
        for field_name in idx.fields:
            if field_name not in seen_names:
                # Add new field, preserving order by adding to end
                self.fields.append(Field(name=field_name, type=None))
                seen_names.add(field_name)

finish_init(db_flavor)

Complete initialization of all edges with vertex configuration.

Source code in graflo/architecture/vertex.py

def finish_init(self, db_flavor: DBFlavor):
    """Complete initialization of all edges with vertex configuration."""
    self.fields = [
        Field(name=f.name, type=FieldType.STRING)
        if f.type is None and db_flavor == DBFlavor.TIGERGRAPH
        else f
        for f in self.fields
    ]

VertexConfig dataclass

Bases: BaseDataclass

Configuration for managing vertices.

This class manages vertices, providing methods for accessing and manipulating vertex configurations.

Attributes:

Name	Type	Description
vertices	list[Vertex]	List of vertex configurations
blank_vertices	list[str]	List of blank vertex names
force_types	dict[str, list]	Dictionary mapping vertex names to type lists
db_flavor	DBFlavor	Database flavor (ARANGO or NEO4J)

Source code in graflo/architecture/vertex.py

@dataclasses.dataclass
class VertexConfig(BaseDataclass):
    """Configuration for managing vertices.

    This class manages vertices, providing methods for accessing
    and manipulating vertex configurations.

    Attributes:
        vertices: List of vertex configurations
        blank_vertices: List of blank vertex names
        force_types: Dictionary mapping vertex names to type lists
        db_flavor: Database flavor (ARANGO or NEO4J)
    """

    vertices: list[Vertex]
    blank_vertices: list[str] = dataclasses.field(default_factory=list)
    force_types: dict[str, list] = dataclasses.field(default_factory=dict)
    db_flavor: DBFlavor = DBFlavor.ARANGO

    def __post_init__(self):
        """Initialize the vertex configuration.

        Creates internal mappings and validates blank vertices.

        Raises:
            ValueError: If blank vertices are not defined in the configuration
        """
        self._vertices_map: dict[str, Vertex] = {
            item.name: item for item in self.vertices
        }

        # TODO replace by types
        # vertex_name -> [numeric fields]
        self._vertex_numeric_fields_map = {}

        if set(self.blank_vertices) - set(self.vertex_set):
            raise ValueError(
                f" Blank vertices {self.blank_vertices} are not defined as vertices"
            )

    @property
    def vertex_set(self):
        """Get set of vertex names.

        Returns:
            set[str]: Set of vertex names
        """
        return set(self._vertices_map.keys())

    @property
    def vertex_list(self):
        """Get list of vertex configurations.

        Returns:
            list[Vertex]: List of vertex configurations
        """
        return list(self._vertices_map.values())

    def _get_vertex_by_name_or_dbname(self, identifier: str) -> Vertex:
        """Get vertex by name or dbname.

        Args:
            identifier: Vertex name or dbname

        Returns:
            Vertex: The vertex object

        Raises:
            KeyError: If vertex is not found by name or dbname
        """
        # First try by name (most common case)
        if identifier in self._vertices_map:
            return self._vertices_map[identifier]

        # Try by dbname
        for vertex in self._vertices_map.values():
            if vertex.dbname == identifier:
                return vertex

        # Not found
        available_names = list(self._vertices_map.keys())
        available_dbnames = [v.dbname for v in self._vertices_map.values()]
        raise KeyError(
            f"Vertex '{identifier}' not found by name or dbname. "
            f"Available names: {available_names}, "
            f"Available dbnames: {available_dbnames}"
        )

    def vertex_dbname(self, vertex_name):
        """Get database name for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            str: Database name for the vertex

        Raises:
            KeyError: If vertex is not found
        """
        try:
            value = self._vertices_map[vertex_name].dbname
        except KeyError as e:
            logger.error(
                "Available vertices :"
                f" {self._vertices_map.keys()}; vertex"
                f" requested : {vertex_name}"
            )
            raise e
        return value

    def index(self, vertex_name) -> Index:
        """Get primary index for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            Index: Primary index for the vertex
        """
        return self._vertices_map[vertex_name].indexes[0]

    def indexes(self, vertex_name) -> list[Index]:
        """Get all indexes for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            list[Index]: List of indexes for the vertex
        """
        return self._vertices_map[vertex_name].indexes

    def fields(self, vertex_name: str) -> list[Field]:
        """Get fields for a vertex.

        Args:
            vertex_name: Name of the vertex or dbname

        Returns:
            list[Field]: List of Field objects
        """
        # Get vertex by name or dbname
        vertex = self._get_vertex_by_name_or_dbname(vertex_name)

        return vertex.fields

    def fields_names(
        self,
        vertex_name: str,
    ) -> list[str]:
        """Get field names for a vertex as strings.

        Args:
            vertex_name: Name of the vertex or dbname

        Returns:
            list[str]: List of field names as strings
        """
        vertex = self._get_vertex_by_name_or_dbname(vertex_name)
        return vertex.field_names

    def numeric_fields_list(self, vertex_name):
        """Get list of numeric fields for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            tuple: Tuple of numeric field names

        Raises:
            ValueError: If vertex is not defined in config
        """
        if vertex_name in self.vertex_set:
            if vertex_name in self._vertex_numeric_fields_map:
                return self._vertex_numeric_fields_map[vertex_name]
            else:
                return ()
        else:
            raise ValueError(
                " Accessing vertex numeric fields: vertex"
                f" {vertex_name} was not defined in config"
            )

    def filters(self, vertex_name) -> list[Expression]:
        """Get filter expressions for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            list[Expression]: List of filter expressions
        """
        if vertex_name in self._vertices_map:
            return self._vertices_map[vertex_name].filters
        else:
            return []

    def update_vertex(self, v: Vertex):
        """Update vertex configuration.

        Args:
            v: Vertex configuration to update
        """
        self._vertices_map[v.name] = v

    def __getitem__(self, key: str):
        """Get vertex configuration by name.

        Args:
            key: Vertex name

        Returns:
            Vertex: Vertex configuration

        Raises:
            KeyError: If vertex is not found
        """
        if key in self._vertices_map:
            return self._vertices_map[key]
        else:
            raise KeyError(f"Vertex {key} absent")

    def __setitem__(self, key: str, value: Vertex):
        """Set vertex configuration by name.

        Args:
            key: Vertex name
            value: Vertex configuration
        """
        self._vertices_map[key] = value

    def finish_init(self, db_flavor: DBFlavor):
        """Complete initialization of all edges with vertex configuration."""

        for v in self.vertices:
            v.finish_init(db_flavor)

vertex_list property

Get list of vertex configurations.

Returns:

Type	Description
	list[Vertex]: List of vertex configurations

vertex_set property

Get set of vertex names.

Returns:

Type	Description
	set[str]: Set of vertex names

__getitem__(key)

Get vertex configuration by name.

Parameters:

Name	Type	Description	Default
key	str	Vertex name	required

Returns:

Name	Type	Description
Vertex		Vertex configuration

Raises:

Type	Description
KeyError	If vertex is not found

Source code in graflo/architecture/vertex.py

def __getitem__(self, key: str):
    """Get vertex configuration by name.

    Args:
        key: Vertex name

    Returns:
        Vertex: Vertex configuration

    Raises:
        KeyError: If vertex is not found
    """
    if key in self._vertices_map:
        return self._vertices_map[key]
    else:
        raise KeyError(f"Vertex {key} absent")

__post_init__()

Initialize the vertex configuration.

Creates internal mappings and validates blank vertices.

Raises:

Type	Description
ValueError	If blank vertices are not defined in the configuration

Source code in graflo/architecture/vertex.py

def __post_init__(self):
    """Initialize the vertex configuration.

    Creates internal mappings and validates blank vertices.

    Raises:
        ValueError: If blank vertices are not defined in the configuration
    """
    self._vertices_map: dict[str, Vertex] = {
        item.name: item for item in self.vertices
    }

    # TODO replace by types
    # vertex_name -> [numeric fields]
    self._vertex_numeric_fields_map = {}

    if set(self.blank_vertices) - set(self.vertex_set):
        raise ValueError(
            f" Blank vertices {self.blank_vertices} are not defined as vertices"
        )

__setitem__(key, value)

Set vertex configuration by name.

Parameters:

Name	Type	Description	Default
key	str	Vertex name	required
value	Vertex	Vertex configuration	required

Source code in graflo/architecture/vertex.py

def __setitem__(self, key: str, value: Vertex):
    """Set vertex configuration by name.

    Args:
        key: Vertex name
        value: Vertex configuration
    """
    self._vertices_map[key] = value

fields(vertex_name)

Get fields for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name	str	Name of the vertex or dbname	required

Returns:

Type	Description
list[Field]	list[Field]: List of Field objects

Source code in graflo/architecture/vertex.py

def fields(self, vertex_name: str) -> list[Field]:
    """Get fields for a vertex.

    Args:
        vertex_name: Name of the vertex or dbname

    Returns:
        list[Field]: List of Field objects
    """
    # Get vertex by name or dbname
    vertex = self._get_vertex_by_name_or_dbname(vertex_name)

    return vertex.fields

fields_names(vertex_name)

Get field names for a vertex as strings.

Parameters:

Name	Type	Description	Default
vertex_name	str	Name of the vertex or dbname	required

Returns:

Type	Description
list[str]	list[str]: List of field names as strings

Source code in graflo/architecture/vertex.py

def fields_names(
    self,
    vertex_name: str,
) -> list[str]:
    """Get field names for a vertex as strings.

    Args:
        vertex_name: Name of the vertex or dbname

    Returns:
        list[str]: List of field names as strings
    """
    vertex = self._get_vertex_by_name_or_dbname(vertex_name)
    return vertex.field_names

filters(vertex_name)

Get filter expressions for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Type	Description
list[Expression]	list[Expression]: List of filter expressions

Source code in graflo/architecture/vertex.py

def filters(self, vertex_name) -> list[Expression]:
    """Get filter expressions for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        list[Expression]: List of filter expressions
    """
    if vertex_name in self._vertices_map:
        return self._vertices_map[vertex_name].filters
    else:
        return []

finish_init(db_flavor)

Complete initialization of all edges with vertex configuration.

Source code in graflo/architecture/vertex.py

def finish_init(self, db_flavor: DBFlavor):
    """Complete initialization of all edges with vertex configuration."""

    for v in self.vertices:
        v.finish_init(db_flavor)

index(vertex_name)

Get primary index for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Name	Type	Description
Index	Index	Primary index for the vertex

Source code in graflo/architecture/vertex.py

def index(self, vertex_name) -> Index:
    """Get primary index for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        Index: Primary index for the vertex
    """
    return self._vertices_map[vertex_name].indexes[0]

indexes(vertex_name)

Get all indexes for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Type	Description
list[Index]	list[Index]: List of indexes for the vertex

Source code in graflo/architecture/vertex.py

def indexes(self, vertex_name) -> list[Index]:
    """Get all indexes for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        list[Index]: List of indexes for the vertex
    """
    return self._vertices_map[vertex_name].indexes

numeric_fields_list(vertex_name)

Get list of numeric fields for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Name	Type	Description
tuple		Tuple of numeric field names

Raises:

Type	Description
ValueError	If vertex is not defined in config

Source code in graflo/architecture/vertex.py

def numeric_fields_list(self, vertex_name):
    """Get list of numeric fields for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        tuple: Tuple of numeric field names

    Raises:
        ValueError: If vertex is not defined in config
    """
    if vertex_name in self.vertex_set:
        if vertex_name in self._vertex_numeric_fields_map:
            return self._vertex_numeric_fields_map[vertex_name]
        else:
            return ()
    else:
        raise ValueError(
            " Accessing vertex numeric fields: vertex"
            f" {vertex_name} was not defined in config"
        )

update_vertex(v)

Update vertex configuration.

Parameters:

Name	Type	Description	Default
v	Vertex	Vertex configuration to update	required

Source code in graflo/architecture/vertex.py

def update_vertex(self, v: Vertex):
    """Update vertex configuration.

    Args:
        v: Vertex configuration to update
    """
    self._vertices_map[v.name] = v

vertex_dbname(vertex_name)

Get database name for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Name	Type	Description
str		Database name for the vertex

Raises:

Type	Description
KeyError	If vertex is not found

Source code in graflo/architecture/vertex.py

def vertex_dbname(self, vertex_name):
    """Get database name for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        str: Database name for the vertex

    Raises:
        KeyError: If vertex is not found
    """
    try:
        value = self._vertices_map[vertex_name].dbname
    except KeyError as e:
        logger.error(
            "Available vertices :"
            f" {self._vertices_map.keys()}; vertex"
            f" requested : {vertex_name}"
        )
        raise e
    return value