graflo.db.conn

Abstract database connection interface for graph databases.

This module defines the abstract interface for database connections, providing a unified API for different graph database implementations. It includes methods for database management, graph structure operations, and data manipulation.

Key Components:

- Connection: Abstract base class for database connections
- ConnectionType: Type variable for connection implementations

The connection interface supports:

- Database/Graph creation and deletion
- Graph structure management (vertex types, edge types)
- Index definition
- Document operations (insert, update, fetch)
- Edge operations
- Aggregation queries

Database Organization Terminology

Different databases organize graph data differently:

• ArangoDB:

  • Database: Top-level container (like a schema)
  • Collections (ArangoDB-specific): Container for vertices (vertex collections)
  • Edge Collections (ArangoDB-specific): Container for edges
  • Graph: Named graph that connects vertex and edge collections

• Neo4j:

  • Database: Top-level container
  • Labels: Categories for nodes (equivalent to vertex types)
  • Relationship Types: Types of relationships (equivalent to edge types)
  • No explicit "graph" concept - all nodes/relationships are in the database

• TigerGraph:

  • Graph: Top-level container (functions like a database in ArangoDB)
  • Vertex Types: Global vertex type definitions (can be shared across graphs)
  • Edge Types: Global edge type definitions (can be shared across graphs)
  • Vertex and edge types are associated with graphs

When using the Connection interface, the terms "vertex type" and "edge type" are used generically to refer to the appropriate concept in each database.

Example

class MyConnection(Connection): ... def create_database(self, name: str): ... # Implementation ... def execute(self, query, **kwargs): ... # Implementation

Connection

Bases: ABC

Abstract base class for database connections.

This class defines the interface that all database connection implementations must follow. It provides methods for database/graph operations, graph structure management (vertex types, edge types), and data manipulation.

Note

All methods marked with @abc.abstractmethod must be implemented by concrete connection classes. Subclasses must set the class attribute flavor to their DBType.

Source code in graflo/db/conn.py

class Connection(abc.ABC):
    """Abstract base class for database connections.

    This class defines the interface that all database connection implementations
    must follow. It provides methods for database/graph operations, graph structure
    management (vertex types, edge types), and data manipulation.

    Note:
        All methods marked with @abc.abstractmethod must be implemented by
        concrete connection classes. Subclasses must set the class attribute
        `flavor` to their DBType.
    """

    flavor: ClassVar[DBType] = DBType.ARANGO  # Overridden by subclasses

    #: Can bulk-export the whole graph (used by export and migration).
    supports_graph_export: ClassVar[bool] = False
    #: Can answer bounded reads — ``fetch_edges`` and therefore traversal.
    #: Distinct from export: reading a neighbourhood is not dumping a graph, and
    #: conflating the two is what limited introspection to three backends.
    supports_graph_read: ClassVar[bool] = True
    #: Has a real :meth:`introspect_graph_schema`, rather than the raising default.
    supports_schema_introspection: ClassVar[bool] = False
    #: Whether :meth:`introspect_graph_schema` samples rows rather than reading a
    #: catalogue. Sampling recovers a *lower bound* — a property present on no
    #: sampled row is simply absent from the result — so a consumer needs this to
    #: know how far to trust the recovered schema. Set ``False`` only where the
    #: backend stores a real DDL catalogue.
    schema_introspection_is_sampled: ClassVar[bool] = True

    def __init__(self):
        """Initialize the connection."""

    @classmethod
    def expression_flavor(cls) -> ExpressionFlavor:
        """Expression flavor for filter rendering (AQL, CYPHER, GSQL).

        Graph connection subclasses must set class attribute `flavor` to a
        DBType present in DB_TYPE_TO_EXPRESSION_FLAVOR.
        """
        return DB_TYPE_TO_EXPRESSION_FLAVOR[cls.flavor]

    @abc.abstractmethod
    def create_database(self, name: str):
        """Create a new database.

        Args:
            name: Name of the database to create
        """

    @abc.abstractmethod
    def delete_database(self, name: str):
        """Delete a database.

        Args:
            name: Name of the database to delete
        """

    @abc.abstractmethod
    def execute(self, query: str | Any, **kwargs: Any) -> Any:
        """Execute a database query.

        Args:
            query: Query to execute
            **kwargs: Additional query parameters

        Returns:
            Query result (database-specific)
        """

    @abc.abstractmethod
    def close(self):
        """Close the database connection."""

    def define_indexes(self, schema: Schema):
        """Define indexes for vertices and edges in the schema.

        Args:
            schema: Schema containing vertex and edge configurations
        """
        self.define_vertex_indexes(schema.core_schema.vertex_config, schema=schema)
        self.define_edge_indexes(
            list(schema.core_schema.edge_config.values()), schema=schema
        )

    @abc.abstractmethod
    def define_schema(self, schema: Schema):
        """Define vertex and edge classes based on the schema.

        Args:
            schema: Schema containing vertex and edge class definitions
        """

    @abc.abstractmethod
    def delete_graph_structure(
        self,
        vertex_types: tuple[str, ...] | list[str] = (),
        graph_names: tuple[str, ...] | list[str] = (),
        delete_all: bool = False,
    ) -> None:
        """Delete graph structure (graphs, vertex types, edge types) from the database.

        This method deletes graphs and their associated vertex/edge types.
        The exact behavior depends on the database implementation:

        - ArangoDB: Deletes graphs and collections (vertex/edge collections)
        - Neo4j: Deletes nodes from labels (vertex types) and relationships
        - TigerGraph: Deletes graphs, vertex types, edge types, and jobs

        Args:
            vertex_types: Vertex type names to delete (database-specific interpretation)
            graph_names: Graph/database names to delete
            delete_all: If True, delete all targeted graph structures.
                This is destructive and should only be used with explicit intent.
        """

    @abc.abstractmethod
    def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
        """Ensure the target graph/database/space namespace exists (op 1).

        Args:
            schema: Schema whose metadata/config resolves the namespace name.
            create: If True, create the namespace when missing (idempotent where
                supported). If False, require an existing namespace or raise
                NamespaceNotFoundError.
        """

    @abc.abstractmethod
    def apply_target_schema(
        self,
        schema: Schema,
        *,
        recreate: bool,
        create_namespace: bool = True,
    ) -> None:
        """Define vertex/edge schema artifacts and indexes (op 2).

        Args:
            schema: Schema to apply.
            recreate: If True, drop existing schema artifacts before defining.
                If False and artifacts already exist, raises SchemaExistsError.
            create_namespace: Whether namespace creation is allowed. Backends use
                this during recreate to decide if the graph/db shell may be dropped.
        """

    @property
    def _reported_edge_directions(self) -> set["EdgeId"]:
        """Edges already reported by :meth:`report_edge_direction_support`.

        Held in ``__dict__`` rather than set in ``__init__`` so every backend
        gets it without touching eight constructors.
        """
        seen = self.__dict__.get("_edge_direction_seen")
        if seen is None:
            seen = set()
            self.__dict__["_edge_direction_seen"] = seen
        return seen

    def edge_direction_diagnostics(
        self, schema: Schema
    ) -> list["EdgeDirectionDiagnostic"]:
        """How this backend will treat each logically undirected edge in ``schema``.

        Returned as data, not just logged, so an API or UI can surface it —
        ``Edge.directed`` is authored far from where its consequences land, and
        a log line reaches nobody driving GraFlo over HTTP. Empty when the
        schema declares no undirected edges, or when the backend represents them
        natively. See :mod:`graflo.db.edge_direction_support`.
        """
        from graflo.db.edge_direction_support import check_schema_edge_directions

        return check_schema_edge_directions(self.flavor, schema)

    def report_edge_direction_support(self, schema: Schema) -> None:
        """Log :meth:`edge_direction_diagnostics` once per distinct edge.

        Backends call this from :meth:`apply_target_schema`, which runs on every
        define/recreate; the diagnostics are a static property of the schema and
        the target, so repeating them each time is noise.
        """
        for diagnostic in self.edge_direction_diagnostics(schema):
            if diagnostic.edge_id in self._reported_edge_directions:
                continue
            self._reported_edge_directions.add(diagnostic.edge_id)
            level = (
                logging.WARNING if diagnostic.severity == "warning" else logging.INFO
            )
            logger.log(level, "%s %s", diagnostic.message, diagnostic.remedy)

    def init_db(
        self,
        schema: Schema,
        recreate_schema: bool = False,
        *,
        create_namespace: bool = True,
    ) -> None:
        """Convenience wrapper: ensure namespace then apply schema.

        Prefer calling ensure_target_namespace and apply_target_schema directly.
        """
        self.ensure_target_namespace(schema, create=create_namespace)
        self.apply_target_schema(
            schema, recreate=recreate_schema, create_namespace=create_namespace
        )

    @abc.abstractmethod
    def clear_data(self, schema: Schema) -> None:
        """Remove all data from the graph without dropping or changing the schema.

        Args:
            schema: Schema describing the graph (used to identify collections/labels).
        """

    @abc.abstractmethod
    def upsert_docs_batch(
        self,
        docs: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        **kwargs: Any,
    ) -> None:
        """Upsert a batch of documents.

        Args:
            docs: Documents to upsert
            class_name: Name of the vertex type (or collection/label in database-specific terms)
            match_keys: Keys to match for upsert
            **kwargs: Additional upsert parameters
        """

    @abc.abstractmethod
    def insert_edges_batch(
        self,
        docs_edges: list[list[dict[str, Any]]] | list[Any] | None,
        source_class: str,
        target_class: str,
        relation_name: str,
        match_keys_source: tuple[str, ...],
        match_keys_target: tuple[str, ...],
        filter_uniques: bool = True,
        head: int | None = None,
        **kwargs: Any,
    ) -> None:
        """Insert a batch of edges.

        Args:
            docs_edges: Edge documents to insert
            source_class: Source vertex type/class
            target_class: Target vertex type/class
            relation_name: Name of the edge type/relation
            match_keys_source: Keys to match source vertices
            match_keys_target: Keys to match target vertices
            filter_uniques: Whether to filter unique edges
            head: Optional limit on number of edges to insert
            **kwargs: Additional insertion parameters (see also
                :func:`consume_insert_edges_kwargs`):
                - dry: If True, do not execute writes (supported where implemented)
                - collection_name: Edge collection (ArangoDB) or unused type-specific name
                - uniq_weight_fields: Uniqueness fields (ArangoDB UPSERT match)
                - uniq_weight_collections: Uniqueness collections (ArangoDB UPSERT)
                - on_duplicate: ArangoDB only. ``\"ignore\"`` (default): ``INSERT`` with
                  ``ignoreErrors``; ``\"upsert\"``: AQL ``UPSERT`` when a matching edge
                  may already exist (align match keys with a unique index).
                - relationship_merge_properties: Property names for Cypher MERGE
                  (Neo4j, FalkorDB, Memgraph) so parallel edges differ by weights
        """

    @abc.abstractmethod
    def insert_return_batch(
        self, docs: list[dict[str, Any]], class_name: str
    ) -> list[dict[str, Any]] | str:
        """Insert documents and return the inserted documents.

        Args:
            docs: Documents to insert
            class_name: Name of the vertex type (or collection/label in database-specific terms)

        Returns:
            list | str: Inserted documents, or a query string (database-specific behavior).
                Most implementations return a list of inserted documents. ArangoDB returns
                an AQL query string for deferred execution.
        """

    @abc.abstractmethod
    def fetch_docs(
        self,
        class_name: str,
        filters: list[Any] | dict[str, Any] | None = None,
        limit: int | None = None,
        return_keys: list[str] | None = None,
        unset_keys: list[str] | None = None,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """Fetch documents from a vertex type.

        Args:
            class_name: Name of the vertex type (or collection/label in database-specific terms)
            filters: Query filters
            limit: Maximum number of documents to return
            return_keys: Keys to return
            unset_keys: Keys to unset
            **kwargs: Additional database-specific parameters (e.g., field_types for TigerGraph)

        Returns:
            list: Fetched documents
        """

    @abc.abstractmethod
    def fetch_edges(
        self,
        from_type: str,
        from_id: str,
        edge_type: str | None = None,
        to_type: str | None = None,
        to_id: str | None = None,
        filters: list[Any] | dict[str, Any] | None = None,
        limit: int | None = None,
        return_keys: list[str] | None = None,
        unset_keys: list[str] | None = None,
        direction: EdgeDirection = EdgeDirection.OUT,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """Fetch edges incident to one vertex.

        ``from_type`` / ``from_id`` name the **anchor** vertex; ``direction``
        decides which orientations are followed from it, and ``to_type`` /
        ``to_id`` constrain the vertex at the *other* end — whichever end that
        is. For a logically undirected edge the correct value is
        :attr:`EdgeDirection.ANY`; :func:`~graflo.db.edge_direction_support.default_direction_for_edge`
        derives it from the schema edge.

        Args:
            from_type: Anchor vertex type
            from_id: Anchor vertex ID (required)
            edge_type: Optional edge type to filter by
            to_type: Optional vertex type of the other endpoint
            to_id: Optional vertex ID of the other endpoint
            filters: Additional query filters
            limit: Maximum number of edges to return
            return_keys: Keys to return (projection)
            unset_keys: Keys to exclude (projection)
            direction: Orientations to follow from the anchor. Defaults to
                :attr:`EdgeDirection.OUT`, the historical behaviour.
            **kwargs: Additional database-specific parameters

        Returns:
            list: List of fetched edges

        Raises:
            UnsupportedEdgeDirectionError: if the backend cannot follow the edge
                in the requested direction (TigerGraph without a reverse type).
        """

    @abc.abstractmethod
    def fetch_present_documents(
        self,
        batch: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        keep_keys: list[str] | tuple[str, ...] | None = None,
        flatten: bool = False,
        filters: list[Any] | dict[str, Any] | None = None,
    ) -> list[dict[str, Any]] | dict[int, list[dict[str, Any]]]:
        """Fetch documents that exist in the database.

        Args:
            batch: Batch of documents to check
            class_name: Name of the collection
            match_keys: Keys to match
            keep_keys: Keys to keep in result
            flatten: Whether to flatten the result. If True, returns a flat list.
                If False, returns a dict mapping batch indices to matching documents.
            filters: Additional query filters

        Returns:
            list | dict: Documents that exist in the database. Returns a list if
                flatten=True, otherwise returns a dict mapping batch indices to documents.
        """

    def resolve_vertices(
        self,
        class_name: str,
        key_docs: list[dict[str, Any]],
        match_keys: tuple[str, ...],
        return_keys: tuple[str, ...],
        *,
        chunk_size: int = DEFAULT_RESOLVE_CHUNK_SIZE,
    ) -> dict[int, list[dict[str, Any]]]:
        """Locate vertices by an arbitrary field-set, preserving multiplicity.

        Used to attach edge endpoints declared by a *secondary identity*: the
        caller passes documents carrying the secondary fields and gets back the
        matching vertices projected onto *return_keys* (the primary identity),
        so the edge write itself stays a plain primary-key operation.

        Unlike :meth:`fetch_present_documents`, every match is returned rather
        than the first, because the caller's ambiguity policy needs the count.

        This default implementation issues one filtered
        :meth:`fetch_docs` per chunk of distinct keys and works on any backend
        whose ``fetch_docs`` honours ``filters``. Backends override it where a
        cheaper or more expressive lookup exists.

        Args:
            class_name: Storage name of the vertex type to search
            key_docs: Documents carrying values for *match_keys*
            match_keys: Field-set to match on (the secondary identity)
            return_keys: Fields to project onto the matched vertices
            chunk_size: Distinct keys per lookup query

        Returns:
            dict: Position in *key_docs* -> every vertex it matched. Positions
                with an unresolvable (partial) key or no match are absent.
        """
        if not key_docs or not match_keys:
            return {}

        keys = distinct_keys(key_docs, match_keys)
        if not keys:
            return {}

        fetch_keys = list(dict.fromkeys([*match_keys, *return_keys]))
        buckets: dict[tuple[Any, ...], list[dict[str, Any]]] = {}
        for chunk in chunked(keys, chunk_size):
            filters = build_match_filter(match_keys, chunk)
            docs = self.fetch_docs(
                class_name,
                filters=filters,
                return_keys=fetch_keys,
            )
            for key, matched in bucket_by_key(list(docs or []), match_keys).items():
                buckets.setdefault(key, []).extend(matched)

        return index_matches_by_doc(key_docs, match_keys, buckets)

    @abc.abstractmethod
    def aggregate(
        self,
        class_name: str,
        aggregation_function: AggregationType,
        discriminant: str | None = None,
        aggregated_field: str | None = None,
        filters: "FilterExpression | list[Any] | dict[str, Any] | None" = None,
    ) -> int | float | list[dict[str, Any]] | dict[str, int | float] | None:
        """Perform aggregation on a collection.

        Args:
            class_name: Name of the collection
            aggregation_function: Type of aggregation to perform
            discriminant: Field to group by
            aggregated_field: Field to aggregate
            filters: Query filters. Accepts a built ``FilterExpression`` as well
                as the shorthand dict/list forms, matching ``fetch_edges`` —
                callers holding a typed query model should not have to round-trip
                it back through a dict to aggregate.

        Returns:
            Aggregation results (type depends on aggregation function)
        """

    @abc.abstractmethod
    def keep_absent_documents(
        self,
        batch: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        keep_keys: list[str] | tuple[str, ...] | None = None,
        filters: list[Any] | dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        """Keep documents that don't exist in the database.

        Args:
            batch: Batch of documents to check
            class_name: Name of the collection
            match_keys: Keys to match
            keep_keys: Keys to keep in result
            filters: Additional query filters

        Returns:
            list: Documents that don't exist in the database
        """

    @abc.abstractmethod
    def define_vertex_indexes(
        self, vertex_config: VertexConfig, schema: Schema | None = None
    ):
        """Define indexes for vertex classes.

        Args:
            vertex_config: Vertex configuration containing index definitions
        """

    @abc.abstractmethod
    def define_edge_indexes(self, edges: list[Edge], schema: Schema | None = None):
        """Define indexes for edge classes.

        Args:
            edges: List of edge configurations containing index definitions
        """

    def define_vertex_classes(self, schema: Schema) -> None:
        """Define vertex classes based on schema.

        This method is called from define_schema() to create vertex types/collections.
        Most implementations take a Schema. Some implementations (like TigerGraph)
        may override with a more specific signature (VertexConfig).

        Default implementation is a no-op. Override in subclasses as needed.

        Args:
            schema: Schema containing vertex definitions
        """

    def define_edge_classes(self, edges: list[Edge]) -> None:
        """Define edge classes based on edge configurations.

        This method is called from define_schema() to create edge types/collections.

        Default implementation is a no-op. Override in subclasses as needed.

        Args:
            edges: List of edge configurations to create
        """

    def bulk_load_begin(
        self, schema: Schema, bulk_cfg: TigergraphBulkLoadConfig
    ) -> str:
        """Start a native bulk-load session (CSV staging + LOADING JOB).

        Raises:
            UnsupportedBulkLoad: For backends that only support REST/document APIs.
        """
        raise UnsupportedBulkLoad(
            f"Database flavor {self.flavor!r} does not support native bulk load"
        )

    def bulk_load_append(
        self, session_id: str, gc: GraphContainer, schema: Schema
    ) -> None:
        """Append one cast batch to the active bulk-load session."""
        raise UnsupportedBulkLoad(
            f"Database flavor {self.flavor!r} does not support native bulk load"
        )

    def bulk_load_finalize(
        self,
        session_id: str,
        schema: Schema,
        *,
        bindings: "Bindings | None" = None,
        connection_provider: "ConnectionProvider | None" = None,
    ) -> str:
        """Close staging files, optionally upload to S3, run LOADING JOB, return GSQL log text."""
        raise UnsupportedBulkLoad(
            f"Database flavor {self.flavor!r} does not support native bulk load"
        )

    def graph_neighbors(
        self,
        vertex_type: str,
        key: str | dict[str, Any],
        *,
        hops: int = 1,
        direction: EdgeDirection = EdgeDirection.OUT,
        edge_types: Sequence[str] | None = None,
        filters: Any | None = None,
        limit: int | None = None,
        schema: Schema | None = None,
    ) -> GraphContainer:
        """Bounded neighbourhood around one anchor vertex.

        Returns a :class:`~graflo.architecture.graph_types.container.GraphContainer`
        so every backend answers in the same shape — the point of the whole read
        path being DB-agnostic.

        The default is a breadth-first composition of :meth:`fetch_edges` and
        :meth:`fetch_docs`, which gives correct multi-hop semantics on any
        backend that can answer a single hop. Backends with a native multi-hop
        form (AQL ``1..k``, Cypher variable-length, nGQL ``GO … STEPS``) override
        this for a single round trip; the conformance suite asserts the override
        returns exactly what the default would.

        Args:
            vertex_type: Logical type of the anchor vertex.
            key: Anchor's identity — a raw id, or a field mapping to resolve.
            hops: Maximum hop distance. Must be >= 1.
            direction: Orientations followed from the anchor. Edges declared
                ``directed=False`` are followed both ways regardless.
            edge_types: Restrict traversal to these logical relation names.
            filters: Optional ``FilterExpression`` applied to edges.
            limit: Maximum edges to accumulate.
            schema: Schema used for logical -> storage name resolution. Required
                for anything but the simplest single-collection graphs.

        Returns:
            GraphContainer: vertices and edges reached, deduplicated.

        Raises:
            UnsupportedEdgeDirectionError: if the backend cannot follow an edge
                in the requested direction (TigerGraph without a reverse type).
        """
        from graflo.db.traversal import bfs_neighbors

        return bfs_neighbors(
            self,
            anchor_type=vertex_type,
            anchor_key=key,
            hops=hops,
            direction=direction,
            edge_types=edge_types,
            filters=filters,
            limit=limit,
            schema=schema,
        )

    def traverse(self, query: Any, *, schema: Schema) -> GraphContainer:
        """Answer a :class:`~graflo.architecture.query.TraverseQuery`.

        The multi-seed form of :meth:`graph_neighbors`. Seeds are walked in
        order and merged into one container, so a vertex reachable from several
        seeds appears once.

        The query is **not** validated here. Cap enforcement belongs to the
        surface that accepted the request, before any connection is opened —
        validating again at the driver would make the ordering ambiguous and
        invite a caller to skip the earlier check.

        Args:
            query: A ``TraverseQuery``. Typed as ``Any`` because ``db`` sits at
                layer 5 and ``architecture.query`` at layer 2; importing it at
                module scope would be an upward import.
            schema: Required, for logical -> storage name resolution.

        Returns:
            GraphContainer: everything reached from any seed, deduplicated.
        """
        container = GraphContainer()
        for seed in query.seeds:
            reached = self.graph_neighbors(
                seed["vertex_type"],
                seed["key"],
                hops=query.max_hops,
                direction=query.edge_direction,
                edge_types=query.edge_relations,
                filters=query.filters,
                limit=query.limit,
                schema=schema,
            )
            for vertex_type, docs in reached.vertices.items():
                container.vertices.setdefault(vertex_type, []).extend(docs)
            for edge_id, rows in reached.edges.items():
                container.edges.setdefault(edge_id, []).extend(rows)
        container.pick_unique()
        return container

    def introspect_graph_schema(
        self,
        schema_name: str | None = None,
        *,
        sample_limit: int = 100,
    ) -> Schema:
        """Infer a graflo :class:`Schema` from this graph database.

        Graph connection subclasses implement sampling-based introspection.
        """
        raise NotImplementedError(
            f"introspect_graph_schema is not implemented for {type(self).__name__}"
        )

    def fetch_all_docs(
        self,
        class_name: str,
        *,
        limit: int | None = None,
    ) -> list[dict[str, Any]]:
        """Fetch all documents for a vertex type/collection."""
        raise NotImplementedError(
            f"fetch_all_docs is not implemented for {type(self).__name__}"
        )

    def fetch_all_edges(
        self,
        source_class: str,
        target_class: str,
        relation_name: str | None,
        *,
        match_keys_source: tuple[str, ...] | None = None,
        match_keys_target: tuple[str, ...] | None = None,
        limit: int | None = None,
        collection_name: str | None = None,
    ) -> list[list[dict[str, Any]]]:
        """Fetch all edges between two vertex types.

        Returns:
            List of ``[source_doc, target_doc, edge_properties]`` triples.
        """
        raise NotImplementedError(
            f"fetch_all_edges is not implemented for {type(self).__name__}"
        )

__init__()

Initialize the connection.

Source code in graflo/db/conn.py

def __init__(self):
    """Initialize the connection."""

aggregate(class_name, aggregation_function, discriminant=None, aggregated_field=None, filters=None) abstractmethod

Perform aggregation on a collection.

Parameters:

Name	Type	Description	Default
class_name	str	Name of the collection	required
aggregation_function	AggregationType	Type of aggregation to perform	required
discriminant	str | None	Field to group by	None
aggregated_field	str | None	Field to aggregate	None
filters	FilterExpression | list[Any] | dict[str, Any] | None	Query filters. Accepts a built FilterExpression as well as the shorthand dict/list forms, matching fetch_edges — callers holding a typed query model should not have to round-trip it back through a dict to aggregate.	None

Returns:

Type	Description
int | float | list[dict[str, Any]] | dict[str, int | float] | None	Aggregation results (type depends on aggregation function)

Source code in graflo/db/conn.py

@abc.abstractmethod
def aggregate(
    self,
    class_name: str,
    aggregation_function: AggregationType,
    discriminant: str | None = None,
    aggregated_field: str | None = None,
    filters: "FilterExpression | list[Any] | dict[str, Any] | None" = None,
) -> int | float | list[dict[str, Any]] | dict[str, int | float] | None:
    """Perform aggregation on a collection.

    Args:
        class_name: Name of the collection
        aggregation_function: Type of aggregation to perform
        discriminant: Field to group by
        aggregated_field: Field to aggregate
        filters: Query filters. Accepts a built ``FilterExpression`` as well
            as the shorthand dict/list forms, matching ``fetch_edges`` —
            callers holding a typed query model should not have to round-trip
            it back through a dict to aggregate.

    Returns:
        Aggregation results (type depends on aggregation function)
    """

apply_target_schema(schema, *, recreate, create_namespace=True) abstractmethod

Define vertex/edge schema artifacts and indexes (op 2).

Parameters:

Name	Type	Description	Default
schema	Schema	Schema to apply.	required
recreate	bool	If True, drop existing schema artifacts before defining. If False and artifacts already exist, raises SchemaExistsError.	required
create_namespace	bool	Whether namespace creation is allowed. Backends use this during recreate to decide if the graph/db shell may be dropped.	True

Source code in graflo/db/conn.py

@abc.abstractmethod
def apply_target_schema(
    self,
    schema: Schema,
    *,
    recreate: bool,
    create_namespace: bool = True,
) -> None:
    """Define vertex/edge schema artifacts and indexes (op 2).

    Args:
        schema: Schema to apply.
        recreate: If True, drop existing schema artifacts before defining.
            If False and artifacts already exist, raises SchemaExistsError.
        create_namespace: Whether namespace creation is allowed. Backends use
            this during recreate to decide if the graph/db shell may be dropped.
    """

bulk_load_append(session_id, gc, schema)

Append one cast batch to the active bulk-load session.

Source code in graflo/db/conn.py

def bulk_load_append(
    self, session_id: str, gc: GraphContainer, schema: Schema
) -> None:
    """Append one cast batch to the active bulk-load session."""
    raise UnsupportedBulkLoad(
        f"Database flavor {self.flavor!r} does not support native bulk load"
    )

bulk_load_begin(schema, bulk_cfg)

Start a native bulk-load session (CSV staging + LOADING JOB).

Raises:

Type	Description
UnsupportedBulkLoad	For backends that only support REST/document APIs.

Source code in graflo/db/conn.py

def bulk_load_begin(
    self, schema: Schema, bulk_cfg: TigergraphBulkLoadConfig
) -> str:
    """Start a native bulk-load session (CSV staging + LOADING JOB).

    Raises:
        UnsupportedBulkLoad: For backends that only support REST/document APIs.
    """
    raise UnsupportedBulkLoad(
        f"Database flavor {self.flavor!r} does not support native bulk load"
    )

bulk_load_finalize(session_id, schema, *, bindings=None, connection_provider=None)

Close staging files, optionally upload to S3, run LOADING JOB, return GSQL log text.

Source code in graflo/db/conn.py

def bulk_load_finalize(
    self,
    session_id: str,
    schema: Schema,
    *,
    bindings: "Bindings | None" = None,
    connection_provider: "ConnectionProvider | None" = None,
) -> str:
    """Close staging files, optionally upload to S3, run LOADING JOB, return GSQL log text."""
    raise UnsupportedBulkLoad(
        f"Database flavor {self.flavor!r} does not support native bulk load"
    )

clear_data(schema) abstractmethod

Remove all data from the graph without dropping or changing the schema.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema describing the graph (used to identify collections/labels).	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def clear_data(self, schema: Schema) -> None:
    """Remove all data from the graph without dropping or changing the schema.

    Args:
        schema: Schema describing the graph (used to identify collections/labels).
    """

close() abstractmethod

Close the database connection.

Source code in graflo/db/conn.py

@abc.abstractmethod
def close(self):
    """Close the database connection."""

create_database(name) abstractmethod

Create a new database.

Parameters:

Name	Type	Description	Default
name	str	Name of the database to create	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def create_database(self, name: str):
    """Create a new database.

    Args:
        name: Name of the database to create
    """

define_edge_classes(edges)

Define edge classes based on edge configurations.

This method is called from define_schema() to create edge types/collections.

Default implementation is a no-op. Override in subclasses as needed.

Parameters:

Name	Type	Description	Default
edges	list[Edge]	List of edge configurations to create	required

Source code in graflo/db/conn.py

def define_edge_classes(self, edges: list[Edge]) -> None:
    """Define edge classes based on edge configurations.

    This method is called from define_schema() to create edge types/collections.

    Default implementation is a no-op. Override in subclasses as needed.

    Args:
        edges: List of edge configurations to create
    """

define_edge_indexes(edges, schema=None) abstractmethod

Define indexes for edge classes.

Parameters:

Name	Type	Description	Default
edges	list[Edge]	List of edge configurations containing index definitions	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def define_edge_indexes(self, edges: list[Edge], schema: Schema | None = None):
    """Define indexes for edge classes.

    Args:
        edges: List of edge configurations containing index definitions
    """

define_indexes(schema)

Define indexes for vertices and edges in the schema.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema containing vertex and edge configurations	required

Source code in graflo/db/conn.py

def define_indexes(self, schema: Schema):
    """Define indexes for vertices and edges in the schema.

    Args:
        schema: Schema containing vertex and edge configurations
    """
    self.define_vertex_indexes(schema.core_schema.vertex_config, schema=schema)
    self.define_edge_indexes(
        list(schema.core_schema.edge_config.values()), schema=schema
    )

define_schema(schema) abstractmethod

Define vertex and edge classes based on the schema.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema containing vertex and edge class definitions	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def define_schema(self, schema: Schema):
    """Define vertex and edge classes based on the schema.

    Args:
        schema: Schema containing vertex and edge class definitions
    """

define_vertex_classes(schema)

Define vertex classes based on schema.

This method is called from define_schema() to create vertex types/collections. Most implementations take a Schema. Some implementations (like TigerGraph) may override with a more specific signature (VertexConfig).

Default implementation is a no-op. Override in subclasses as needed.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema containing vertex definitions	required

Source code in graflo/db/conn.py

def define_vertex_classes(self, schema: Schema) -> None:
    """Define vertex classes based on schema.

    This method is called from define_schema() to create vertex types/collections.
    Most implementations take a Schema. Some implementations (like TigerGraph)
    may override with a more specific signature (VertexConfig).

    Default implementation is a no-op. Override in subclasses as needed.

    Args:
        schema: Schema containing vertex definitions
    """

define_vertex_indexes(vertex_config, schema=None) abstractmethod

Define indexes for vertex classes.

Parameters:

Name	Type	Description	Default
vertex_config	VertexConfig	Vertex configuration containing index definitions	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def define_vertex_indexes(
    self, vertex_config: VertexConfig, schema: Schema | None = None
):
    """Define indexes for vertex classes.

    Args:
        vertex_config: Vertex configuration containing index definitions
    """

delete_database(name) abstractmethod

Delete a database.

Parameters:

Name	Type	Description	Default
name	str	Name of the database to delete	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def delete_database(self, name: str):
    """Delete a database.

    Args:
        name: Name of the database to delete
    """

delete_graph_structure(vertex_types=(), graph_names=(), delete_all=False) abstractmethod

Delete graph structure (graphs, vertex types, edge types) from the database.

This method deletes graphs and their associated vertex/edge types. The exact behavior depends on the database implementation:

• ArangoDB: Deletes graphs and collections (vertex/edge collections)
• Neo4j: Deletes nodes from labels (vertex types) and relationships
• TigerGraph: Deletes graphs, vertex types, edge types, and jobs

Parameters:

Name	Type	Description	Default
vertex_types	tuple[str, ...] | list[str]	Vertex type names to delete (database-specific interpretation)	()
graph_names	tuple[str, ...] | list[str]	Graph/database names to delete	()
delete_all	bool	If True, delete all targeted graph structures. This is destructive and should only be used with explicit intent.	False

Source code in graflo/db/conn.py

@abc.abstractmethod
def delete_graph_structure(
    self,
    vertex_types: tuple[str, ...] | list[str] = (),
    graph_names: tuple[str, ...] | list[str] = (),
    delete_all: bool = False,
) -> None:
    """Delete graph structure (graphs, vertex types, edge types) from the database.

    This method deletes graphs and their associated vertex/edge types.
    The exact behavior depends on the database implementation:

    - ArangoDB: Deletes graphs and collections (vertex/edge collections)
    - Neo4j: Deletes nodes from labels (vertex types) and relationships
    - TigerGraph: Deletes graphs, vertex types, edge types, and jobs

    Args:
        vertex_types: Vertex type names to delete (database-specific interpretation)
        graph_names: Graph/database names to delete
        delete_all: If True, delete all targeted graph structures.
            This is destructive and should only be used with explicit intent.
    """

edge_direction_diagnostics(schema)

How this backend will treat each logically undirected edge in schema.

Returned as data, not just logged, so an API or UI can surface it — Edge.directed is authored far from where its consequences land, and a log line reaches nobody driving GraFlo over HTTP. Empty when the schema declares no undirected edges, or when the backend represents them natively. See :mod:graflo.db.edge_direction_support.

Source code in graflo/db/conn.py

def edge_direction_diagnostics(
    self, schema: Schema
) -> list["EdgeDirectionDiagnostic"]:
    """How this backend will treat each logically undirected edge in ``schema``.

    Returned as data, not just logged, so an API or UI can surface it —
    ``Edge.directed`` is authored far from where its consequences land, and
    a log line reaches nobody driving GraFlo over HTTP. Empty when the
    schema declares no undirected edges, or when the backend represents them
    natively. See :mod:`graflo.db.edge_direction_support`.
    """
    from graflo.db.edge_direction_support import check_schema_edge_directions

    return check_schema_edge_directions(self.flavor, schema)

ensure_target_namespace(schema, *, create) abstractmethod

Ensure the target graph/database/space namespace exists (op 1).

Parameters:

Name	Type	Description	Default
schema	Schema	Schema whose metadata/config resolves the namespace name.	required
create	bool	If True, create the namespace when missing (idempotent where supported). If False, require an existing namespace or raise NamespaceNotFoundError.	required

Source code in graflo/db/conn.py

@abc.abstractmethod
def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
    """Ensure the target graph/database/space namespace exists (op 1).

    Args:
        schema: Schema whose metadata/config resolves the namespace name.
        create: If True, create the namespace when missing (idempotent where
            supported). If False, require an existing namespace or raise
            NamespaceNotFoundError.
    """

execute(query, **kwargs) abstractmethod

Execute a database query.

Parameters:

Name	Type	Description	Default
query	str | Any	Query to execute	required
**kwargs	Any	Additional query parameters	{}

Returns:

Type	Description
Any	Query result (database-specific)

Source code in graflo/db/conn.py

@abc.abstractmethod
def execute(self, query: str | Any, **kwargs: Any) -> Any:
    """Execute a database query.

    Args:
        query: Query to execute
        **kwargs: Additional query parameters

    Returns:
        Query result (database-specific)
    """

expression_flavor() classmethod

Expression flavor for filter rendering (AQL, CYPHER, GSQL).

Graph connection subclasses must set class attribute flavor to a DBType present in DB_TYPE_TO_EXPRESSION_FLAVOR.

Source code in graflo/db/conn.py

@classmethod
def expression_flavor(cls) -> ExpressionFlavor:
    """Expression flavor for filter rendering (AQL, CYPHER, GSQL).

    Graph connection subclasses must set class attribute `flavor` to a
    DBType present in DB_TYPE_TO_EXPRESSION_FLAVOR.
    """
    return DB_TYPE_TO_EXPRESSION_FLAVOR[cls.flavor]

fetch_all_docs(class_name, *, limit=None)

Fetch all documents for a vertex type/collection.

Source code in graflo/db/conn.py

def fetch_all_docs(
    self,
    class_name: str,
    *,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """Fetch all documents for a vertex type/collection."""
    raise NotImplementedError(
        f"fetch_all_docs is not implemented for {type(self).__name__}"
    )

fetch_all_edges(source_class, target_class, relation_name, *, match_keys_source=None, match_keys_target=None, limit=None, collection_name=None)

Fetch all edges between two vertex types.

Returns:

Type	Description
list[list[dict[str, Any]]]	List of [source_doc, target_doc, edge_properties] triples.

Source code in graflo/db/conn.py

def fetch_all_edges(
    self,
    source_class: str,
    target_class: str,
    relation_name: str | None,
    *,
    match_keys_source: tuple[str, ...] | None = None,
    match_keys_target: tuple[str, ...] | None = None,
    limit: int | None = None,
    collection_name: str | None = None,
) -> list[list[dict[str, Any]]]:
    """Fetch all edges between two vertex types.

    Returns:
        List of ``[source_doc, target_doc, edge_properties]`` triples.
    """
    raise NotImplementedError(
        f"fetch_all_edges is not implemented for {type(self).__name__}"
    )

fetch_docs(class_name, filters=None, limit=None, return_keys=None, unset_keys=None, **kwargs) abstractmethod

Fetch documents from a vertex type.

Parameters:

Name	Type	Description	Default
class_name	str	Name of the vertex type (or collection/label in database-specific terms)	required
filters	list[Any] | dict[str, Any] | None	Query filters	None
limit	int | None	Maximum number of documents to return	None
return_keys	list[str] | None	Keys to return	None
unset_keys	list[str] | None	Keys to unset	None
**kwargs	Any	Additional database-specific parameters (e.g., field_types for TigerGraph)	{}

Returns:

Name	Type	Description
list	list[dict[str, Any]]	Fetched documents

Source code in graflo/db/conn.py

@abc.abstractmethod
def fetch_docs(
    self,
    class_name: str,
    filters: list[Any] | dict[str, Any] | None = None,
    limit: int | None = None,
    return_keys: list[str] | None = None,
    unset_keys: list[str] | None = None,
    **kwargs: Any,
) -> list[dict[str, Any]]:
    """Fetch documents from a vertex type.

    Args:
        class_name: Name of the vertex type (or collection/label in database-specific terms)
        filters: Query filters
        limit: Maximum number of documents to return
        return_keys: Keys to return
        unset_keys: Keys to unset
        **kwargs: Additional database-specific parameters (e.g., field_types for TigerGraph)

    Returns:
        list: Fetched documents
    """

fetch_edges(from_type, from_id, edge_type=None, to_type=None, to_id=None, filters=None, limit=None, return_keys=None, unset_keys=None, direction=EdgeDirection.OUT, **kwargs) abstractmethod

Fetch edges incident to one vertex.

from_type / from_id name the anchor vertex; direction decides which orientations are followed from it, and to_type / to_id constrain the vertex at the other end — whichever end that is. For a logically undirected edge the correct value is :attr:EdgeDirection.ANY; :func:~graflo.db.edge_direction_support.default_direction_for_edge derives it from the schema edge.

Parameters:

Name	Type	Description	Default
from_type	str	Anchor vertex type	required
from_id	str	Anchor vertex ID (required)	required
edge_type	str | None	Optional edge type to filter by	None
to_type	str | None	Optional vertex type of the other endpoint	None
to_id	str | None	Optional vertex ID of the other endpoint	None
filters	list[Any] | dict[str, Any] | None	Additional query filters	None
limit	int | None	Maximum number of edges to return	None
return_keys	list[str] | None	Keys to return (projection)	None
unset_keys	list[str] | None	Keys to exclude (projection)	None
direction	EdgeDirection	Orientations to follow from the anchor. Defaults to :attr:EdgeDirection.OUT, the historical behaviour.	OUT
**kwargs	Any	Additional database-specific parameters	{}

Returns:

Name	Type	Description
list	list[dict[str, Any]]	List of fetched edges

Raises:

Type	Description
UnsupportedEdgeDirectionError	if the backend cannot follow the edge in the requested direction (TigerGraph without a reverse type).

Source code in graflo/db/conn.py

@abc.abstractmethod
def fetch_edges(
    self,
    from_type: str,
    from_id: str,
    edge_type: str | None = None,
    to_type: str | None = None,
    to_id: str | None = None,
    filters: list[Any] | dict[str, Any] | None = None,
    limit: int | None = None,
    return_keys: list[str] | None = None,
    unset_keys: list[str] | None = None,
    direction: EdgeDirection = EdgeDirection.OUT,
    **kwargs: Any,
) -> list[dict[str, Any]]:
    """Fetch edges incident to one vertex.

    ``from_type`` / ``from_id`` name the **anchor** vertex; ``direction``
    decides which orientations are followed from it, and ``to_type`` /
    ``to_id`` constrain the vertex at the *other* end — whichever end that
    is. For a logically undirected edge the correct value is
    :attr:`EdgeDirection.ANY`; :func:`~graflo.db.edge_direction_support.default_direction_for_edge`
    derives it from the schema edge.

    Args:
        from_type: Anchor vertex type
        from_id: Anchor vertex ID (required)
        edge_type: Optional edge type to filter by
        to_type: Optional vertex type of the other endpoint
        to_id: Optional vertex ID of the other endpoint
        filters: Additional query filters
        limit: Maximum number of edges to return
        return_keys: Keys to return (projection)
        unset_keys: Keys to exclude (projection)
        direction: Orientations to follow from the anchor. Defaults to
            :attr:`EdgeDirection.OUT`, the historical behaviour.
        **kwargs: Additional database-specific parameters

    Returns:
        list: List of fetched edges

    Raises:
        UnsupportedEdgeDirectionError: if the backend cannot follow the edge
            in the requested direction (TigerGraph without a reverse type).
    """

fetch_present_documents(batch, class_name, match_keys, keep_keys=None, flatten=False, filters=None) abstractmethod

Fetch documents that exist in the database.

Parameters:

Name	Type	Description	Default
batch	list[dict[str, Any]]	Batch of documents to check	required
class_name	str	Name of the collection	required
match_keys	list[str] | tuple[str, ...]	Keys to match	required
keep_keys	list[str] | tuple[str, ...] | None	Keys to keep in result	None
flatten	bool	Whether to flatten the result. If True, returns a flat list. If False, returns a dict mapping batch indices to matching documents.	False
filters	list[Any] | dict[str, Any] | None	Additional query filters	None

Returns:

Type	Description
list[dict[str, Any]] | dict[int, list[dict[str, Any]]]	list | dict: Documents that exist in the database. Returns a list if flatten=True, otherwise returns a dict mapping batch indices to documents.

Source code in graflo/db/conn.py

@abc.abstractmethod
def fetch_present_documents(
    self,
    batch: list[dict[str, Any]],
    class_name: str,
    match_keys: list[str] | tuple[str, ...],
    keep_keys: list[str] | tuple[str, ...] | None = None,
    flatten: bool = False,
    filters: list[Any] | dict[str, Any] | None = None,
) -> list[dict[str, Any]] | dict[int, list[dict[str, Any]]]:
    """Fetch documents that exist in the database.

    Args:
        batch: Batch of documents to check
        class_name: Name of the collection
        match_keys: Keys to match
        keep_keys: Keys to keep in result
        flatten: Whether to flatten the result. If True, returns a flat list.
            If False, returns a dict mapping batch indices to matching documents.
        filters: Additional query filters

    Returns:
        list | dict: Documents that exist in the database. Returns a list if
            flatten=True, otherwise returns a dict mapping batch indices to documents.
    """

graph_neighbors(vertex_type, key, *, hops=1, direction=EdgeDirection.OUT, edge_types=None, filters=None, limit=None, schema=None)

Bounded neighbourhood around one anchor vertex.

Returns a :class:~graflo.architecture.graph_types.container.GraphContainer so every backend answers in the same shape — the point of the whole read path being DB-agnostic.

The default is a breadth-first composition of :meth:fetch_edges and :meth:fetch_docs, which gives correct multi-hop semantics on any backend that can answer a single hop. Backends with a native multi-hop form (AQL 1..k, Cypher variable-length, nGQL GO … STEPS) override this for a single round trip; the conformance suite asserts the override returns exactly what the default would.

Parameters:

Name	Type	Description	Default
vertex_type	str	Logical type of the anchor vertex.	required
key	str | dict[str, Any]	Anchor's identity — a raw id, or a field mapping to resolve.	required
hops	int	Maximum hop distance. Must be >= 1.	1
direction	EdgeDirection	Orientations followed from the anchor. Edges declared directed=False are followed both ways regardless.	OUT
edge_types	Sequence[str] | None	Restrict traversal to these logical relation names.	None
filters	Any | None	Optional FilterExpression applied to edges.	None
limit	int | None	Maximum edges to accumulate.	None
schema	Schema | None	Schema used for logical -> storage name resolution. Required for anything but the simplest single-collection graphs.	None

Returns:

Name	Type	Description
GraphContainer	GraphContainer	vertices and edges reached, deduplicated.

Raises:

Type	Description
UnsupportedEdgeDirectionError	if the backend cannot follow an edge in the requested direction (TigerGraph without a reverse type).

Source code in graflo/db/conn.py

def graph_neighbors(
    self,
    vertex_type: str,
    key: str | dict[str, Any],
    *,
    hops: int = 1,
    direction: EdgeDirection = EdgeDirection.OUT,
    edge_types: Sequence[str] | None = None,
    filters: Any | None = None,
    limit: int | None = None,
    schema: Schema | None = None,
) -> GraphContainer:
    """Bounded neighbourhood around one anchor vertex.

    Returns a :class:`~graflo.architecture.graph_types.container.GraphContainer`
    so every backend answers in the same shape — the point of the whole read
    path being DB-agnostic.

    The default is a breadth-first composition of :meth:`fetch_edges` and
    :meth:`fetch_docs`, which gives correct multi-hop semantics on any
    backend that can answer a single hop. Backends with a native multi-hop
    form (AQL ``1..k``, Cypher variable-length, nGQL ``GO … STEPS``) override
    this for a single round trip; the conformance suite asserts the override
    returns exactly what the default would.

    Args:
        vertex_type: Logical type of the anchor vertex.
        key: Anchor's identity — a raw id, or a field mapping to resolve.
        hops: Maximum hop distance. Must be >= 1.
        direction: Orientations followed from the anchor. Edges declared
            ``directed=False`` are followed both ways regardless.
        edge_types: Restrict traversal to these logical relation names.
        filters: Optional ``FilterExpression`` applied to edges.
        limit: Maximum edges to accumulate.
        schema: Schema used for logical -> storage name resolution. Required
            for anything but the simplest single-collection graphs.

    Returns:
        GraphContainer: vertices and edges reached, deduplicated.

    Raises:
        UnsupportedEdgeDirectionError: if the backend cannot follow an edge
            in the requested direction (TigerGraph without a reverse type).
    """
    from graflo.db.traversal import bfs_neighbors

    return bfs_neighbors(
        self,
        anchor_type=vertex_type,
        anchor_key=key,
        hops=hops,
        direction=direction,
        edge_types=edge_types,
        filters=filters,
        limit=limit,
        schema=schema,
    )

init_db(schema, recreate_schema=False, *, create_namespace=True)

Convenience wrapper: ensure namespace then apply schema.

Prefer calling ensure_target_namespace and apply_target_schema directly.

Source code in graflo/db/conn.py

def init_db(
    self,
    schema: Schema,
    recreate_schema: bool = False,
    *,
    create_namespace: bool = True,
) -> None:
    """Convenience wrapper: ensure namespace then apply schema.

    Prefer calling ensure_target_namespace and apply_target_schema directly.
    """
    self.ensure_target_namespace(schema, create=create_namespace)
    self.apply_target_schema(
        schema, recreate=recreate_schema, create_namespace=create_namespace
    )

insert_edges_batch(docs_edges, source_class, target_class, relation_name, match_keys_source, match_keys_target, filter_uniques=True, head=None, **kwargs) abstractmethod

Insert a batch of edges.

Parameters:

Name	Type	Description	Default
docs_edges	list[list[dict[str, Any]]] | list[Any] | None	Edge documents to insert	required
source_class	str	Source vertex type/class	required
target_class	str	Target vertex type/class	required
relation_name	str	Name of the edge type/relation	required
match_keys_source	tuple[str, ...]	Keys to match source vertices	required
match_keys_target	tuple[str, ...]	Keys to match target vertices	required
filter_uniques	bool	Whether to filter unique edges	True
head	int | None	Optional limit on number of edges to insert	None
**kwargs	Any	Additional insertion parameters (see also :func:consume_insert_edges_kwargs): - dry: If True, do not execute writes (supported where implemented) - collection_name: Edge collection (ArangoDB) or unused type-specific name - uniq_weight_fields: Uniqueness fields (ArangoDB UPSERT match) - uniq_weight_collections: Uniqueness collections (ArangoDB UPSERT) - on_duplicate: ArangoDB only. "ignore" (default): INSERT with ignoreErrors; "upsert": AQL UPSERT when a matching edge may already exist (align match keys with a unique index). - relationship_merge_properties: Property names for Cypher MERGE (Neo4j, FalkorDB, Memgraph) so parallel edges differ by weights	{}

Source code in graflo/db/conn.py

@abc.abstractmethod
def insert_edges_batch(
    self,
    docs_edges: list[list[dict[str, Any]]] | list[Any] | None,
    source_class: str,
    target_class: str,
    relation_name: str,
    match_keys_source: tuple[str, ...],
    match_keys_target: tuple[str, ...],
    filter_uniques: bool = True,
    head: int | None = None,
    **kwargs: Any,
) -> None:
    """Insert a batch of edges.

    Args:
        docs_edges: Edge documents to insert
        source_class: Source vertex type/class
        target_class: Target vertex type/class
        relation_name: Name of the edge type/relation
        match_keys_source: Keys to match source vertices
        match_keys_target: Keys to match target vertices
        filter_uniques: Whether to filter unique edges
        head: Optional limit on number of edges to insert
        **kwargs: Additional insertion parameters (see also
            :func:`consume_insert_edges_kwargs`):
            - dry: If True, do not execute writes (supported where implemented)
            - collection_name: Edge collection (ArangoDB) or unused type-specific name
            - uniq_weight_fields: Uniqueness fields (ArangoDB UPSERT match)
            - uniq_weight_collections: Uniqueness collections (ArangoDB UPSERT)
            - on_duplicate: ArangoDB only. ``\"ignore\"`` (default): ``INSERT`` with
              ``ignoreErrors``; ``\"upsert\"``: AQL ``UPSERT`` when a matching edge
              may already exist (align match keys with a unique index).
            - relationship_merge_properties: Property names for Cypher MERGE
              (Neo4j, FalkorDB, Memgraph) so parallel edges differ by weights
    """

insert_return_batch(docs, class_name) abstractmethod

Insert documents and return the inserted documents.

Parameters:

Name	Type	Description	Default
docs	list[dict[str, Any]]	Documents to insert	required
class_name	str	Name of the vertex type (or collection/label in database-specific terms)	required

Returns:

Type	Description
list[dict[str, Any]] | str	list | str: Inserted documents, or a query string (database-specific behavior). Most implementations return a list of inserted documents. ArangoDB returns an AQL query string for deferred execution.

Source code in graflo/db/conn.py

@abc.abstractmethod
def insert_return_batch(
    self, docs: list[dict[str, Any]], class_name: str
) -> list[dict[str, Any]] | str:
    """Insert documents and return the inserted documents.

    Args:
        docs: Documents to insert
        class_name: Name of the vertex type (or collection/label in database-specific terms)

    Returns:
        list | str: Inserted documents, or a query string (database-specific behavior).
            Most implementations return a list of inserted documents. ArangoDB returns
            an AQL query string for deferred execution.
    """

introspect_graph_schema(schema_name=None, *, sample_limit=100)

Infer a graflo :class:Schema from this graph database.

Graph connection subclasses implement sampling-based introspection.

Source code in graflo/db/conn.py

def introspect_graph_schema(
    self,
    schema_name: str | None = None,
    *,
    sample_limit: int = 100,
) -> Schema:
    """Infer a graflo :class:`Schema` from this graph database.

    Graph connection subclasses implement sampling-based introspection.
    """
    raise NotImplementedError(
        f"introspect_graph_schema is not implemented for {type(self).__name__}"
    )

keep_absent_documents(batch, class_name, match_keys, keep_keys=None, filters=None) abstractmethod

Keep documents that don't exist in the database.

Parameters:

Name	Type	Description	Default
batch	list[dict[str, Any]]	Batch of documents to check	required
class_name	str	Name of the collection	required
match_keys	list[str] | tuple[str, ...]	Keys to match	required
keep_keys	list[str] | tuple[str, ...] | None	Keys to keep in result	None
filters	list[Any] | dict[str, Any] | None	Additional query filters	None

Returns:

Name	Type	Description
list	list[dict[str, Any]]	Documents that don't exist in the database

Source code in graflo/db/conn.py

@abc.abstractmethod
def keep_absent_documents(
    self,
    batch: list[dict[str, Any]],
    class_name: str,
    match_keys: list[str] | tuple[str, ...],
    keep_keys: list[str] | tuple[str, ...] | None = None,
    filters: list[Any] | dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Keep documents that don't exist in the database.

    Args:
        batch: Batch of documents to check
        class_name: Name of the collection
        match_keys: Keys to match
        keep_keys: Keys to keep in result
        filters: Additional query filters

    Returns:
        list: Documents that don't exist in the database
    """

report_edge_direction_support(schema)

Log :meth:edge_direction_diagnostics once per distinct edge.

Backends call this from :meth:apply_target_schema, which runs on every define/recreate; the diagnostics are a static property of the schema and the target, so repeating them each time is noise.

Source code in graflo/db/conn.py

def report_edge_direction_support(self, schema: Schema) -> None:
    """Log :meth:`edge_direction_diagnostics` once per distinct edge.

    Backends call this from :meth:`apply_target_schema`, which runs on every
    define/recreate; the diagnostics are a static property of the schema and
    the target, so repeating them each time is noise.
    """
    for diagnostic in self.edge_direction_diagnostics(schema):
        if diagnostic.edge_id in self._reported_edge_directions:
            continue
        self._reported_edge_directions.add(diagnostic.edge_id)
        level = (
            logging.WARNING if diagnostic.severity == "warning" else logging.INFO
        )
        logger.log(level, "%s %s", diagnostic.message, diagnostic.remedy)

resolve_vertices(class_name, key_docs, match_keys, return_keys, *, chunk_size=DEFAULT_RESOLVE_CHUNK_SIZE)

Locate vertices by an arbitrary field-set, preserving multiplicity.

Used to attach edge endpoints declared by a secondary identity: the caller passes documents carrying the secondary fields and gets back the matching vertices projected onto return_keys (the primary identity), so the edge write itself stays a plain primary-key operation.

Unlike :meth:fetch_present_documents, every match is returned rather than the first, because the caller's ambiguity policy needs the count.

This default implementation issues one filtered :meth:fetch_docs per chunk of distinct keys and works on any backend whose fetch_docs honours filters. Backends override it where a cheaper or more expressive lookup exists.

Parameters:

Name	Type	Description	Default
class_name	str	Storage name of the vertex type to search	required
key_docs	list[dict[str, Any]]	Documents carrying values for match_keys	required
match_keys	tuple[str, ...]	Field-set to match on (the secondary identity)	required
return_keys	tuple[str, ...]	Fields to project onto the matched vertices	required
chunk_size	int	Distinct keys per lookup query	DEFAULT_RESOLVE_CHUNK_SIZE

Returns:

Name	Type	Description
dict	dict[int, list[dict[str, Any]]]	Position in key_docs -> every vertex it matched. Positions with an unresolvable (partial) key or no match are absent.

Source code in graflo/db/conn.py

def resolve_vertices(
    self,
    class_name: str,
    key_docs: list[dict[str, Any]],
    match_keys: tuple[str, ...],
    return_keys: tuple[str, ...],
    *,
    chunk_size: int = DEFAULT_RESOLVE_CHUNK_SIZE,
) -> dict[int, list[dict[str, Any]]]:
    """Locate vertices by an arbitrary field-set, preserving multiplicity.

    Used to attach edge endpoints declared by a *secondary identity*: the
    caller passes documents carrying the secondary fields and gets back the
    matching vertices projected onto *return_keys* (the primary identity),
    so the edge write itself stays a plain primary-key operation.

    Unlike :meth:`fetch_present_documents`, every match is returned rather
    than the first, because the caller's ambiguity policy needs the count.

    This default implementation issues one filtered
    :meth:`fetch_docs` per chunk of distinct keys and works on any backend
    whose ``fetch_docs`` honours ``filters``. Backends override it where a
    cheaper or more expressive lookup exists.

    Args:
        class_name: Storage name of the vertex type to search
        key_docs: Documents carrying values for *match_keys*
        match_keys: Field-set to match on (the secondary identity)
        return_keys: Fields to project onto the matched vertices
        chunk_size: Distinct keys per lookup query

    Returns:
        dict: Position in *key_docs* -> every vertex it matched. Positions
            with an unresolvable (partial) key or no match are absent.
    """
    if not key_docs or not match_keys:
        return {}

    keys = distinct_keys(key_docs, match_keys)
    if not keys:
        return {}

    fetch_keys = list(dict.fromkeys([*match_keys, *return_keys]))
    buckets: dict[tuple[Any, ...], list[dict[str, Any]]] = {}
    for chunk in chunked(keys, chunk_size):
        filters = build_match_filter(match_keys, chunk)
        docs = self.fetch_docs(
            class_name,
            filters=filters,
            return_keys=fetch_keys,
        )
        for key, matched in bucket_by_key(list(docs or []), match_keys).items():
            buckets.setdefault(key, []).extend(matched)

    return index_matches_by_doc(key_docs, match_keys, buckets)

traverse(query, *, schema)

Answer a :class:~graflo.architecture.query.TraverseQuery.

The multi-seed form of :meth:graph_neighbors. Seeds are walked in order and merged into one container, so a vertex reachable from several seeds appears once.

The query is not validated here. Cap enforcement belongs to the surface that accepted the request, before any connection is opened — validating again at the driver would make the ordering ambiguous and invite a caller to skip the earlier check.

Parameters:

Name	Type	Description	Default
query	Any	A TraverseQuery. Typed as Any because db sits at layer 5 and architecture.query at layer 2; importing it at module scope would be an upward import.	required
schema	Schema	Required, for logical -> storage name resolution.	required

Returns:

Name	Type	Description
GraphContainer	GraphContainer	everything reached from any seed, deduplicated.

Source code in graflo/db/conn.py

def traverse(self, query: Any, *, schema: Schema) -> GraphContainer:
    """Answer a :class:`~graflo.architecture.query.TraverseQuery`.

    The multi-seed form of :meth:`graph_neighbors`. Seeds are walked in
    order and merged into one container, so a vertex reachable from several
    seeds appears once.

    The query is **not** validated here. Cap enforcement belongs to the
    surface that accepted the request, before any connection is opened —
    validating again at the driver would make the ordering ambiguous and
    invite a caller to skip the earlier check.

    Args:
        query: A ``TraverseQuery``. Typed as ``Any`` because ``db`` sits at
            layer 5 and ``architecture.query`` at layer 2; importing it at
            module scope would be an upward import.
        schema: Required, for logical -> storage name resolution.

    Returns:
        GraphContainer: everything reached from any seed, deduplicated.
    """
    container = GraphContainer()
    for seed in query.seeds:
        reached = self.graph_neighbors(
            seed["vertex_type"],
            seed["key"],
            hops=query.max_hops,
            direction=query.edge_direction,
            edge_types=query.edge_relations,
            filters=query.filters,
            limit=query.limit,
            schema=schema,
        )
        for vertex_type, docs in reached.vertices.items():
            container.vertices.setdefault(vertex_type, []).extend(docs)
        for edge_id, rows in reached.edges.items():
            container.edges.setdefault(edge_id, []).extend(rows)
    container.pick_unique()
    return container

upsert_docs_batch(docs, class_name, match_keys, **kwargs) abstractmethod

Upsert a batch of documents.

Parameters:

Name	Type	Description	Default
docs	list[dict[str, Any]]	Documents to upsert	required
class_name	str	Name of the vertex type (or collection/label in database-specific terms)	required
match_keys	list[str] | tuple[str, ...]	Keys to match for upsert	required
**kwargs	Any	Additional upsert parameters	{}

Source code in graflo/db/conn.py

@abc.abstractmethod
def upsert_docs_batch(
    self,
    docs: list[dict[str, Any]],
    class_name: str,
    match_keys: list[str] | tuple[str, ...],
    **kwargs: Any,
) -> None:
    """Upsert a batch of documents.

    Args:
        docs: Documents to upsert
        class_name: Name of the vertex type (or collection/label in database-specific terms)
        match_keys: Keys to match for upsert
        **kwargs: Additional upsert parameters
    """

ConnectionCapability

Bases: Enum

A capability a caller needs from a connection, by ClassVar name.

Values are the attribute names on :class:Connection so that opening a connection for a purpose and declaring support for that purpose cannot drift apart -- there is one name, not a flag and a matching-by-convention check.

Source code in graflo/db/conn.py

class ConnectionCapability(Enum):
    """A capability a caller needs from a connection, by ClassVar name.

    Values are the attribute names on :class:`Connection` so that opening a
    connection for a purpose and declaring support for that purpose cannot drift
    apart -- there is one name, not a flag and a matching-by-convention check.
    """

    GRAPH_READ = "supports_graph_read"
    GRAPH_EXPORT = "supports_graph_export"
    SCHEMA_INTROSPECTION = "supports_schema_introspection"

    @property
    def label(self) -> str:
        """Human-readable name, for error messages."""
        return {
            ConnectionCapability.GRAPH_READ: "bounded graph reads",
            ConnectionCapability.GRAPH_EXPORT: "bulk graph export",
            ConnectionCapability.SCHEMA_INTROSPECTION: "schema introspection",
        }[self]

label property

Human-readable name, for error messages.

InsertEdgesKwArgs dataclass

Keyword arguments shared by :meth:Connection.insert_edges_batch implementations.

Source code in graflo/db/conn.py

@dataclass(frozen=True)
class InsertEdgesKwArgs:
    """Keyword arguments shared by :meth:`Connection.insert_edges_batch` implementations."""

    dry: bool
    collection_name: str | None
    uniq_weight_fields: Any
    uniq_weight_collections: Any
    on_duplicate: Literal["upsert", "ignore"]
    relationship_merge_properties: Any

NamespaceNotFoundError

Bases: RuntimeError

Raised when create=False and the target graph/database/space does not exist.

Source code in graflo/db/conn.py

class NamespaceNotFoundError(RuntimeError):
    """Raised when create=False and the target graph/database/space does not exist."""

SchemaExistsError

Bases: RuntimeError

Raised when schema artifacts already exist and recreate is False.

Set recreate=True in apply_target_schema (or recreate_schema=True in init_db) to replace the existing schema, or use clear_data=True before ingestion to only clear data without touching the schema.

Source code in graflo/db/conn.py

class SchemaExistsError(RuntimeError):
    """Raised when schema artifacts already exist and recreate is False.

    Set recreate=True in apply_target_schema (or recreate_schema=True in init_db)
    to replace the existing schema, or use clear_data=True before ingestion to
    only clear data without touching the schema.
    """

consume_insert_edges_kwargs(kwargs)

Pop standard insert_edges_batch keys from kwargs and warn on unknown keys.

Mutates kwargs in place (removes consumed keys). Callers should not pass additional keyword arguments beyond those documented on :meth:Connection.insert_edges_batch.

Source code in graflo/db/conn.py

def consume_insert_edges_kwargs(kwargs: dict[str, Any]) -> InsertEdgesKwArgs:
    """Pop standard ``insert_edges_batch`` keys from *kwargs* and warn on unknown keys.

    Mutates *kwargs* in place (removes consumed keys). Callers should not pass
    additional keyword arguments beyond those documented on
    :meth:`Connection.insert_edges_batch`.
    """
    result = InsertEdgesKwArgs(
        dry=bool(kwargs.pop("dry", False)),
        collection_name=kwargs.pop("collection_name", None),
        uniq_weight_fields=kwargs.pop("uniq_weight_fields", None),
        uniq_weight_collections=kwargs.pop("uniq_weight_collections", None),
        on_duplicate=_parse_on_duplicate(kwargs.pop("on_duplicate", "ignore")),
        relationship_merge_properties=kwargs.pop("relationship_merge_properties", None),
    )
    if kwargs:
        logger.warning(
            "insert_edges_batch: unsupported keyword arguments ignored: %s",
            sorted(kwargs.keys()),
        )
        kwargs.clear()
    return result