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

Attributes

ConnectionType = TypeVar('ConnectionType', bound='Connection') module-attribute

DEFAULT_DELETE_CHUNK_SIZE = 1000 module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

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
    #: Applies a ``guard`` to every vertex and edge a traversal reads, every hop.
    #: A backend without it refuses a guard rather than reading unguarded.
    supports_read_guard: ClassVar[bool] = False
    #: Writes ``merge: union`` list fields as the union of the stored and the
    #: written value. A writer refuses such a field on a backend without it.
    supports_union_merge: ClassVar[bool] = False
    #: :meth:`fetch_docs` takes an ``ANY_IN`` filter on a list field. Lookup by a
    #: derived keys field refuses a backend without it.
    supports_membership_read: ClassVar[bool] = False
    #: A migration emitter exists for this backend, so a non-dry-run apply can
    #: be attempted at all. Declaring it is what lets a caller refuse an
    #: unsupported backend **by name, before opening a connection**, instead of
    #: surfacing a driver error from deep inside an emitter.
    #:
    #: It says an emitter is registered — not that every operation is
    #: implemented, and not that applying to a *populated* target works
    #: (it does not, on any backend: the emitters' DDL meets the first-write
    #: guard, which refuses a populated target).
    supports_schema_ddl: 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
    #: Implements :meth:`delete_vertices` and :meth:`delete_edges`.
    supports_instance_delete: ClassVar[bool] = False
    #: Implements :meth:`delete_scoped_vertices` and :meth:`delete_scoped_edges`.
    supports_scoped_delete: ClassVar[bool] = False
    #: Implements the bulk-load protocol (:mod:`graflo.db.bulk`): ``bulk_begin``
    #: through ``bulk_finish``.
    supports_bulk_load: ClassVar[bool] = False
    #: Its bulk writer takes dense integer ids on every vertex and edge endpoint.
    bulk_dense_ids: ClassVar[bool] = False
    #: Its bulk writer locks both endpoints of an edge it creates, so the edges
    #: written at once must share no endpoint: each edge type is then written in
    #: rounds of chunks split by a hash of their endpoints' keys.
    bulk_partition_edges: ClassVar[bool] = False
    #: Implements :mod:`graflo.db.sync`: replacing what changed owner records wrote.
    supports_sync: ClassVar[bool] = False

    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 reset_after_use(self) -> None:
        """End whatever a unit of work left open, before the connection is reused.

        A no-op unless the backend holds per-connection state between calls,
        such as an open transaction.
        """

    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.
        """

    # -- bulk load (graflo.db.bulk) -------------------------------------------

    def bulk_begin(
        self,
        schema: Schema,
        cfg: BulkConfig,
        *,
        recreate: bool = False,
        resume: bool = False,
    ) -> BulkPlan:
        """Start a bulk load into an empty target: namespace and schema DDL.

        The schema is applied through the first-write guard, so a target that
        already holds one is refused unless *recreate*. On *resume* the load
        began before: the guard is skipped and DDL is applied only where missing.

        Raises:
            NotImplementedError: The backend has no bulk writer.
            SchemaExistsError: The target holds a schema and *recreate* is off.
        """
        raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

    def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
        """Create the chunk's vertices; none exists yet unless ``chunk.replay``."""
        raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

    def bulk_prepare_edges(self, schema: Schema) -> None:
        """Make edge endpoints quick to find, once every vertex is written."""
        raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

    def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
        """Create the chunk's edges between existing vertices."""
        raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

    def bulk_finish(self, schema: Schema) -> BulkCounts:
        """Build what the load deferred; return the elements the target holds."""
        raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

    # -- record sync (graflo.db.sync) -------------------------------------------

    def sync_begin(self, schema: Schema, edge_types: list[SyncEdgeType]) -> None:
        """Index every vertex identity, and the owner property of *edge_types*.

        Raises:
            NotImplementedError: The backend has no sync writer.
        """
        raise NotImplementedError(f"{type(self).__name__} has no sync writer")

    def sync_ensure_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
        """Create each vertex whose key is absent, from its row; leave the others."""
        raise NotImplementedError(f"{type(self).__name__} has no sync writer")

    def sync_write_owners(self, schema: Schema, chunk: SyncOwnerChunk) -> SyncWritten:
        """Rewrite the chunk's owners whose stored hash differs, as one unit."""
        raise NotImplementedError(f"{type(self).__name__} has no sync writer")

    @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)
                - relationship_merge_properties: Property names for Cypher MERGE
                  (Neo4j, FalkorDB, Memgraph) so parallel edges differ by weights;
                  on PostgreSQL, the edge table's key columns besides the endpoints
                - union_fields: List properties written as the union of the
                  stored list and the written one; only a backend that declares
                  ``supports_union_merge`` is passed them
        """

    @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. Two are part of
                the contract, passed by every schema-aware caller and ignored
                by backends that do not need them: ``native_inverse_type``
                (``str | None`` — the paired type the database maintains for
                the edge's relation, see ``DatabaseProfile.native_inverse_of``)
                and ``edge_is_undirected`` (``bool`` — the edge type was created
                undirected). A backend whose reverse reachability is decided at
                schema time answers ``IN`` / ``ANY`` from them; rows come back in
                the orientation of the edge as declared, whichever type answered.

        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,
        guard: FilterExpression | None = None,
    ) -> 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
            guard: A trusted predicate every match must satisfy, ANDed with
                the key match; a vertex it hides is never matched.

        Returns:
            dict: Position in *key_docs* -> every vertex it matched. Positions
                with an unresolvable (partial) key or no match are absent.
        """
        if guard is not None:
            self.require_read_guard()
        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 batched(keys, chunk_size):
            match = build_match_filter(match_keys, chunk)
            # Unchanged without a guard; with one, only a backend that declares
            # guard support gets here, and its fetch_docs parses an expression.
            filters: Any = match if guard is None else conjoin(match, guard)
            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)

    def delete_vertices(
        self,
        class_name: str,
        key_docs: list[dict[str, Any]],
        match_keys: tuple[str, ...],
        *,
        chunk_size: int = DEFAULT_DELETE_CHUNK_SIZE,
    ) -> None:
        """Remove vertices, with every edge incident to them.

        Names are stored names, as for every write.

        Args:
            class_name: Stored name of the vertex type.
            key_docs: One document per vertex to remove, carrying *match_keys*.
                A document missing a key field removes nothing.
            match_keys: The vertex type's identity fields, as written.
            chunk_size: Documents per statement.

        Raises:
            NotImplementedError: On a backend that does not declare
                ``supports_instance_delete``.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support deleting vertices"
        )

    def delete_edges(
        self,
        source_class: str,
        target_class: str,
        relation_name: str | None,
        endpoints: list[tuple[dict[str, Any], dict[str, Any]]],
        match_keys_source: tuple[str, ...],
        match_keys_target: tuple[str, ...],
        *,
        collection_name: str | None = None,
        chunk_size: int = DEFAULT_DELETE_CHUNK_SIZE,
    ) -> None:
        """Remove the edges of one type between the given endpoint pairs.

        Every edge of the type from the source to the target of a pair goes,
        whatever its properties. Names are stored names, as for every write.

        Args:
            source_class: Stored name of the source vertex type.
            target_class: Stored name of the target vertex type.
            relation_name: Stored relation, as passed to :meth:`insert_edges_batch`.
            endpoints: ``(source, target)`` documents carrying the match keys.
            match_keys_source: Identity fields of the source type.
            match_keys_target: Identity fields of the target type.
            collection_name: Edge collection, where the backend has one.
            chunk_size: Pairs per statement.

        Raises:
            NotImplementedError: On a backend that does not declare
                ``supports_instance_delete``.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support deleting edges"
        )

    def delete_scoped_vertices(self, class_name: str, field: str, value: Any) -> None:
        """Remove the vertices of one type whose *field* equals *value*.

        Every edge incident to them goes too. Names are stored names.

        Args:
            class_name: Stored name of the vertex type.
            field: Stored name of the scope field.
            value: The scope; never ``None``.

        Raises:
            NotImplementedError: On a backend that does not declare
                ``supports_scoped_delete``.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support deleting vertices by scope"
        )

    def delete_scoped_edges(
        self,
        source_class: str,
        target_class: str,
        relation_name: str | None,
        field: str,
        value: Any,
        *,
        collection_name: str | None = None,
    ) -> None:
        """Remove the edges of one type whose *field* equals *value*.

        Names are stored names, as for :meth:`delete_edges`.

        Args:
            source_class: Stored name of the source vertex type.
            target_class: Stored name of the target vertex type.
            relation_name: Stored relation, as passed to :meth:`insert_edges_batch`.
            field: Stored name of the scope field on the edge.
            value: The scope; never ``None``.
            collection_name: Edge collection, where the backend has one.

        Raises:
            NotImplementedError: On a backend that does not declare
                ``supports_scoped_delete``.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support deleting edges by scope"
        )

    @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 vertex_address(
        self, doc: dict[str, Any], identity_fields: Sequence[str]
    ) -> str | None:
        """How this backend addresses *doc* in an edge query.

        Most backends key a vertex on a single value, so the first identity
        field that is present is the address. NebulaGraph composes a VID from
        *every* identity field, so it overrides this — without which a
        composite identity resolves to a prefix that matches no edge, and the
        traversal returns empty rather than failing.
        """
        for field in identity_fields:
            value = doc.get(field)
            if value is not None:
                return str(value)
        return None

    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,
        guard: FilterExpression | 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.

        Names are logical on both sides: the anchor key and edge filters name
        logical properties, and returned vertex documents and edge rows are keyed
        by them, whatever names the database stores (``db_profile``'s property
        maps, plus any the target flavor needs filled in). Backends implement
        :meth:`_graph_neighbors`, which sees stored names only.

        The default :meth:`_graph_neighbors` 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. ArangoDB (AQL
        ``1..k``) and the Cypher backends (variable-length paths) override it
        for a single round trip; the conformance suite asserts the override
        returns exactly what the default would. NebulaGraph, TigerGraph and
        PostgreSQL use the default.

        Args:
            vertex_type: Logical type of the anchor vertex.
            key: A complete key of the anchor
                (:mod:`~graflo.architecture.schema.anchor`): the value of a
                one-field primary identity, or a mapping naming exactly the
                primary identity, a derived key's source fields, or one secondary
                identity. A primary key is not looked up: a vertex that does not
                exist has no neighbours. Without *schema*, a bare value is the
                stored id.
            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.
            guard: A trusted predicate, never taken from a request, that every
                vertex and edge on a path must satisfy, the anchor included.
                It names fields as stored, and reserved fields may appear in it.

        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).
            ReservedFieldError: if *filters* name a reserved field.
            AnchorKeyError: if *key* is not a complete key, or a secondary
                identity names several vertices or none.
            NotImplementedError: if a *guard* is given and the backend does not
                declare ``supports_read_guard``.
        """
        guarded = self._guard_kwargs(guard)
        if schema is None:
            if isinstance(key, dict):
                from graflo.architecture.schema.anchor import AnchorNotAKeyError

                raise AnchorNotAKeyError(
                    f"an anchor mapping for {vertex_type!r} needs the schema that "
                    "says which fields are keys",
                    vertex_type=vertex_type,
                )
            return self._graph_neighbors(
                vertex_type,
                key,
                hops=hops,
                direction=direction,
                edge_types=edge_types,
                filters=filters,
                limit=limit,
                schema=None,
                **guarded,
            )
        stored, keys = self._stored_view(schema)
        _refuse_reserved(_reserved_names(schema, stored), filters=filters, key=key)
        anchor = self._anchor_identity(schema, vertex_type, key, guard)
        reached = self._graph_neighbors(
            vertex_type,
            keys.anchor_key(vertex_type, anchor),
            hops=hops,
            direction=direction,
            edge_types=edge_types,
            filters=keys.edge_filter(filters, _walked_edge_ids(schema, edge_types)),
            limit=limit,
            schema=stored,
            **guarded,
        )
        return keys.logical_container(reached, schema.core_schema.edge_config)

    def require_read_guard(self) -> None:
        """Refuse a guarded read on a backend that cannot apply the guard.

        Raises:
            NotImplementedError: unless the backend declares ``supports_read_guard``.
        """
        if not self.supports_read_guard:
            raise NotImplementedError(
                f"{type(self).__name__} cannot apply a read guard; a guarded read "
                "is refused rather than answered unguarded"
            )

    def _anchor_identity(
        self,
        schema: Schema,
        vertex_type: str,
        key: str | dict[str, Any],
        guard: FilterExpression | None,
    ) -> str | dict[str, Any]:
        """The primary identity *key* names, in logical names; a bare value as-is.

        Backends then match a complete key only, so no backend can anchor on an
        arbitrary vertex among several that share a non-key value.
        """
        # Deferred: hq sits above db; a secondary identity is looked up through it.
        from graflo.hq.anchor import anchor_identity

        return anchor_identity(schema, vertex_type, key, db=self, guard=guard)

    def _guard_kwargs(self, guard: FilterExpression | None) -> dict[str, Any]:
        """``{"guard": guard}`` for :meth:`_graph_neighbors`, or nothing.

        Passed only when set, so a backend hook that predates guards keeps its
        signature; such a backend never declares ``supports_read_guard``.
        """
        if guard is None:
            return {}
        self.require_read_guard()
        return {"guard": guard}

    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:
        """Backend hook for :meth:`graph_neighbors`, in stored names.

        *schema* is the schema as the database stores it
        (:func:`~graflo.architecture.evolution.sanitize.physical_schema`), and
        *key* and *filters* are already translated to it. The default walks
        breadth-first over :meth:`fetch_edges` and :meth:`fetch_docs`; a backend
        that declares ``supports_read_guard`` and keeps this default overrides
        it to pass ``guard`` on to
        :func:`~graflo.db.traversal.bfs_neighbors`.
        """
        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 _stored_view(self, schema: Schema) -> tuple[Schema, PhysicalKeys]:
        """*schema* as this database stores it, and the key translation to it.

        With nothing stored under another name this is the schema itself (or a
        copy carrying filled-in storage names) and a no-op translation, so the
        common read pays for neither a deep copy nor a document rebuild.
        """
        from graflo.architecture.evolution.sanitize import (
            materialize_physical_schema,
            with_physical_names,
        )

        named = with_physical_names(schema, self.flavor)
        keys = PhysicalKeys(named.db_profile)
        return (materialize_physical_schema(named) if keys.active else named), keys

    def traverse(
        self,
        query: Any,
        *,
        schema: Schema,
        guard: FilterExpression | None = None,
    ) -> 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. Names are logical in and out, as for
        :meth:`graph_neighbors`; the stored view is resolved once for all seeds.

        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.
            guard: As for :meth:`graph_neighbors`.

        Returns:
            GraphContainer: everything reached from any seed, deduplicated.

        Raises:
            ReservedFieldError: if the query names a reserved field.
            AnchorKeyError: if a seed's key is not a complete key, or names
                several vertices or none; ``seed`` is its position.
            NotImplementedError: if a *guard* is given and the backend does not
                declare ``supports_read_guard``.
        """
        guarded = self._guard_kwargs(guard)
        stored, keys = self._stored_view(schema)
        reserved = _reserved_names(schema, stored)
        query.refuse_reserved(reserved)
        for seed in query.seeds:
            _refuse_reserved(reserved, key=seed.get("key"))
        filters = keys.edge_filter(
            query.filters, _walked_edge_ids(schema, query.edge_relations)
        )
        from graflo.architecture.schema.anchor import AnchorKeyError

        # Every seed is resolved before any is walked: one that names no vertex
        # fails the request rather than shrinking its answer.
        anchors: list[tuple[str, str | dict[str, Any]]] = []
        for position, seed in enumerate(query.seeds):
            try:
                anchor = self._anchor_identity(
                    schema, seed["vertex_type"], seed["key"], guard
                )
            except AnchorKeyError as error:
                error.seed = position
                raise
            anchors.append((seed["vertex_type"], anchor))
        container = GraphContainer()
        for vertex_type, anchor in anchors:
            reached = self._graph_neighbors(
                vertex_type,
                keys.anchor_key(vertex_type, anchor),
                hops=query.max_hops,
                direction=query.edge_direction,
                edge_types=query.edge_relations,
                filters=filters,
                limit=query.limit,
                schema=stored,
                **guarded,
            )
            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 keys.logical_container(container, schema.core_schema.edge_config)

    def export_graph_container(
        self, schema: Schema, *, limit: int | None = None
    ) -> GraphContainer:
        """Every vertex and edge of *schema*, read from this database.

        *schema* is the one :meth:`introspect_graph_schema` returned. The
        default reads each vertex type with :meth:`fetch_all_docs` and each
        edge with :meth:`fetch_all_edges`.

        Args:
            schema: The graph to read.
            limit: At most this many records per vertex type and per edge.
        """
        vertices: dict[str, list] = {}
        edges: dict[tuple[str, str, str | None], list] = {}
        vc = schema.core_schema.vertex_config

        for vertex in vc.vertices:
            docs = self.fetch_all_docs(vertex.name, limit=limit)
            if docs:
                vertices[vertex.name] = docs

        for edge in schema.core_schema.edge_config.values():
            # Backends that store endpoints as a single identity value (rather
            # than a resolvable document reference) need to be told which field
            # that value belongs to, so the exported endpoint doc is keyed the
            # way DBWriter resolves it.
            edge_docs = self.fetch_all_edges(
                edge.source,
                edge.target,
                edge.relation,
                match_keys_source=tuple(vc.identity_fields(edge.source)),
                match_keys_target=tuple(vc.identity_fields(edge.target)),
                limit=limit,
            )
            if edge_docs:
                edges[edge.edge_id] = edge_docs

        return GraphContainer(vertices=vertices, edges=edges, linear=[])

    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__}"
        )

Attributes

bulk_dense_ids = False class-attribute

bulk_partition_edges = False class-attribute

flavor = DBType.ARANGO class-attribute

schema_introspection_is_sampled = True class-attribute

supports_bulk_load = False class-attribute

supports_graph_export = False class-attribute

supports_graph_read = True class-attribute

supports_instance_delete = False class-attribute

supports_membership_read = False class-attribute

supports_read_guard = False class-attribute

supports_schema_ddl = False class-attribute

supports_schema_introspection = False class-attribute

supports_scoped_delete = False class-attribute

supports_sync = False class-attribute

supports_union_merge = False class-attribute

Methods:

__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_begin(schema, cfg, *, recreate=False, resume=False)

Start a bulk load into an empty target: namespace and schema DDL.

The schema is applied through the first-write guard, so a target that already holds one is refused unless recreate. On resume the load began before: the guard is skipped and DDL is applied only where missing.

Raises:

Type	Description
NotImplementedError	The backend has no bulk writer.
SchemaExistsError	The target holds a schema and recreate is off.

Source code in graflo/db/conn.py

def bulk_begin(
    self,
    schema: Schema,
    cfg: BulkConfig,
    *,
    recreate: bool = False,
    resume: bool = False,
) -> BulkPlan:
    """Start a bulk load into an empty target: namespace and schema DDL.

    The schema is applied through the first-write guard, so a target that
    already holds one is refused unless *recreate*. On *resume* the load
    began before: the guard is skipped and DDL is applied only where missing.

    Raises:
        NotImplementedError: The backend has no bulk writer.
        SchemaExistsError: The target holds a schema and *recreate* is off.
    """
    raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

bulk_finish(schema)

Build what the load deferred; return the elements the target holds.

Source code in graflo/db/conn.py

def bulk_finish(self, schema: Schema) -> BulkCounts:
    """Build what the load deferred; return the elements the target holds."""
    raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

bulk_prepare_edges(schema)

Make edge endpoints quick to find, once every vertex is written.

Source code in graflo/db/conn.py

def bulk_prepare_edges(self, schema: Schema) -> None:
    """Make edge endpoints quick to find, once every vertex is written."""
    raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

bulk_write_edges(schema, chunk)

Create the chunk's edges between existing vertices.

Source code in graflo/db/conn.py

def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
    """Create the chunk's edges between existing vertices."""
    raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

bulk_write_vertices(schema, chunk)

Create the chunk's vertices; none exists yet unless chunk.replay.

Source code in graflo/db/conn.py

def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
    """Create the chunk's vertices; none exists yet unless ``chunk.replay``."""
    raise NotImplementedError(f"{type(self).__name__} has no bulk writer")

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_edges(source_class, target_class, relation_name, endpoints, match_keys_source, match_keys_target, *, collection_name=None, chunk_size=DEFAULT_DELETE_CHUNK_SIZE)

Remove the edges of one type between the given endpoint pairs.

Every edge of the type from the source to the target of a pair goes, whatever its properties. Names are stored names, as for every write.

Parameters:

Name	Type	Description	Default
source_class	str	Stored name of the source vertex type.	required
target_class	str	Stored name of the target vertex type.	required
relation_name	str | None	Stored relation, as passed to :meth:insert_edges_batch.	required
endpoints	list[tuple[dict[str, Any], dict[str, Any]]]	(source, target) documents carrying the match keys.	required
match_keys_source	tuple[str, ...]	Identity fields of the source type.	required
match_keys_target	tuple[str, ...]	Identity fields of the target type.	required
collection_name	str | None	Edge collection, where the backend has one.	None
chunk_size	int	Pairs per statement.	DEFAULT_DELETE_CHUNK_SIZE

Raises:

Type	Description
NotImplementedError	On a backend that does not declare supports_instance_delete.

Source code in graflo/db/conn.py

def delete_edges(
    self,
    source_class: str,
    target_class: str,
    relation_name: str | None,
    endpoints: list[tuple[dict[str, Any], dict[str, Any]]],
    match_keys_source: tuple[str, ...],
    match_keys_target: tuple[str, ...],
    *,
    collection_name: str | None = None,
    chunk_size: int = DEFAULT_DELETE_CHUNK_SIZE,
) -> None:
    """Remove the edges of one type between the given endpoint pairs.

    Every edge of the type from the source to the target of a pair goes,
    whatever its properties. Names are stored names, as for every write.

    Args:
        source_class: Stored name of the source vertex type.
        target_class: Stored name of the target vertex type.
        relation_name: Stored relation, as passed to :meth:`insert_edges_batch`.
        endpoints: ``(source, target)`` documents carrying the match keys.
        match_keys_source: Identity fields of the source type.
        match_keys_target: Identity fields of the target type.
        collection_name: Edge collection, where the backend has one.
        chunk_size: Pairs per statement.

    Raises:
        NotImplementedError: On a backend that does not declare
            ``supports_instance_delete``.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support deleting edges"
    )

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.
    """

delete_scoped_edges(source_class, target_class, relation_name, field, value, *, collection_name=None)

Remove the edges of one type whose field equals value.

Names are stored names, as for :meth:delete_edges.

Parameters:

Name	Type	Description	Default
source_class	str	Stored name of the source vertex type.	required
target_class	str	Stored name of the target vertex type.	required
relation_name	str | None	Stored relation, as passed to :meth:insert_edges_batch.	required
field	str	Stored name of the scope field on the edge.	required
value	Any	The scope; never None.	required
collection_name	str | None	Edge collection, where the backend has one.	None

Raises:

Type	Description
NotImplementedError	On a backend that does not declare supports_scoped_delete.

Source code in graflo/db/conn.py

def delete_scoped_edges(
    self,
    source_class: str,
    target_class: str,
    relation_name: str | None,
    field: str,
    value: Any,
    *,
    collection_name: str | None = None,
) -> None:
    """Remove the edges of one type whose *field* equals *value*.

    Names are stored names, as for :meth:`delete_edges`.

    Args:
        source_class: Stored name of the source vertex type.
        target_class: Stored name of the target vertex type.
        relation_name: Stored relation, as passed to :meth:`insert_edges_batch`.
        field: Stored name of the scope field on the edge.
        value: The scope; never ``None``.
        collection_name: Edge collection, where the backend has one.

    Raises:
        NotImplementedError: On a backend that does not declare
            ``supports_scoped_delete``.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support deleting edges by scope"
    )

delete_scoped_vertices(class_name, field, value)

Remove the vertices of one type whose field equals value.

Every edge incident to them goes too. Names are stored names.

Parameters:

Name	Type	Description	Default
class_name	str	Stored name of the vertex type.	required
field	str	Stored name of the scope field.	required
value	Any	The scope; never None.	required

Raises:

Type	Description
NotImplementedError	On a backend that does not declare supports_scoped_delete.

Source code in graflo/db/conn.py

def delete_scoped_vertices(self, class_name: str, field: str, value: Any) -> None:
    """Remove the vertices of one type whose *field* equals *value*.

    Every edge incident to them goes too. Names are stored names.

    Args:
        class_name: Stored name of the vertex type.
        field: Stored name of the scope field.
        value: The scope; never ``None``.

    Raises:
        NotImplementedError: On a backend that does not declare
            ``supports_scoped_delete``.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support deleting vertices by scope"
    )

delete_vertices(class_name, key_docs, match_keys, *, chunk_size=DEFAULT_DELETE_CHUNK_SIZE)

Remove vertices, with every edge incident to them.

Names are stored names, as for every write.

Parameters:

Name	Type	Description	Default
class_name	str	Stored name of the vertex type.	required
key_docs	list[dict[str, Any]]	One document per vertex to remove, carrying match_keys. A document missing a key field removes nothing.	required
match_keys	tuple[str, ...]	The vertex type's identity fields, as written.	required
chunk_size	int	Documents per statement.	DEFAULT_DELETE_CHUNK_SIZE

Raises:

Type	Description
NotImplementedError	On a backend that does not declare supports_instance_delete.

Source code in graflo/db/conn.py

def delete_vertices(
    self,
    class_name: str,
    key_docs: list[dict[str, Any]],
    match_keys: tuple[str, ...],
    *,
    chunk_size: int = DEFAULT_DELETE_CHUNK_SIZE,
) -> None:
    """Remove vertices, with every edge incident to them.

    Names are stored names, as for every write.

    Args:
        class_name: Stored name of the vertex type.
        key_docs: One document per vertex to remove, carrying *match_keys*.
            A document missing a key field removes nothing.
        match_keys: The vertex type's identity fields, as written.
        chunk_size: Documents per statement.

    Raises:
        NotImplementedError: On a backend that does not declare
            ``supports_instance_delete``.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support deleting vertices"
    )

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)
    """

export_graph_container(schema, *, limit=None)

Every vertex and edge of schema, read from this database.

schema is the one :meth:introspect_graph_schema returned. The default reads each vertex type with :meth:fetch_all_docs and each edge with :meth:fetch_all_edges.

Parameters:

Name	Type	Description	Default
schema	Schema	The graph to read.	required
limit	int | None	At most this many records per vertex type and per edge.	None

Source code in graflo/db/conn.py

def export_graph_container(
    self, schema: Schema, *, limit: int | None = None
) -> GraphContainer:
    """Every vertex and edge of *schema*, read from this database.

    *schema* is the one :meth:`introspect_graph_schema` returned. The
    default reads each vertex type with :meth:`fetch_all_docs` and each
    edge with :meth:`fetch_all_edges`.

    Args:
        schema: The graph to read.
        limit: At most this many records per vertex type and per edge.
    """
    vertices: dict[str, list] = {}
    edges: dict[tuple[str, str, str | None], list] = {}
    vc = schema.core_schema.vertex_config

    for vertex in vc.vertices:
        docs = self.fetch_all_docs(vertex.name, limit=limit)
        if docs:
            vertices[vertex.name] = docs

    for edge in schema.core_schema.edge_config.values():
        # Backends that store endpoints as a single identity value (rather
        # than a resolvable document reference) need to be told which field
        # that value belongs to, so the exported endpoint doc is keyed the
        # way DBWriter resolves it.
        edge_docs = self.fetch_all_edges(
            edge.source,
            edge.target,
            edge.relation,
            match_keys_source=tuple(vc.identity_fields(edge.source)),
            match_keys_target=tuple(vc.identity_fields(edge.target)),
            limit=limit,
        )
        if edge_docs:
            edges[edge.edge_id] = edge_docs

    return GraphContainer(vertices=vertices, edges=edges, linear=[])

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. Two are part of the contract, passed by every schema-aware caller and ignored by backends that do not need them: native_inverse_type (str | None — the paired type the database maintains for the edge's relation, see DatabaseProfile.native_inverse_of) and edge_is_undirected (bool — the edge type was created undirected). A backend whose reverse reachability is decided at schema time answers IN / ANY from them; rows come back in the orientation of the edge as declared, whichever type answered.	{}

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. Two are part of
            the contract, passed by every schema-aware caller and ignored
            by backends that do not need them: ``native_inverse_type``
            (``str | None`` — the paired type the database maintains for
            the edge's relation, see ``DatabaseProfile.native_inverse_of``)
            and ``edge_is_undirected`` (``bool`` — the edge type was created
            undirected). A backend whose reverse reachability is decided at
            schema time answers ``IN`` / ``ANY`` from them; rows come back in
            the orientation of the edge as declared, whichever type answered.

    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, guard=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.

Names are logical on both sides: the anchor key and edge filters name logical properties, and returned vertex documents and edge rows are keyed by them, whatever names the database stores (db_profile's property maps, plus any the target flavor needs filled in). Backends implement :meth:_graph_neighbors, which sees stored names only.

The default :meth:_graph_neighbors 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. ArangoDB (AQL 1..k) and the Cypher backends (variable-length paths) override it for a single round trip; the conformance suite asserts the override returns exactly what the default would. NebulaGraph, TigerGraph and PostgreSQL use the default.

Parameters:

Name	Type	Description	Default
vertex_type	str	Logical type of the anchor vertex.	required
key	str | dict[str, Any]	A complete key of the anchor (:mod:~graflo.architecture.schema.anchor): the value of a one-field primary identity, or a mapping naming exactly the primary identity, a derived key's source fields, or one secondary identity. A primary key is not looked up: a vertex that does not exist has no neighbours. Without schema, a bare value is the stored id.	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
guard	FilterExpression | None	A trusted predicate, never taken from a request, that every vertex and edge on a path must satisfy, the anchor included. It names fields as stored, and reserved fields may appear in it.	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).
ReservedFieldError	if filters name a reserved field.
AnchorKeyError	if key is not a complete key, or a secondary identity names several vertices or none.
NotImplementedError	if a guard is given and the backend does not declare supports_read_guard.

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,
    guard: FilterExpression | 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.

    Names are logical on both sides: the anchor key and edge filters name
    logical properties, and returned vertex documents and edge rows are keyed
    by them, whatever names the database stores (``db_profile``'s property
    maps, plus any the target flavor needs filled in). Backends implement
    :meth:`_graph_neighbors`, which sees stored names only.

    The default :meth:`_graph_neighbors` 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. ArangoDB (AQL
    ``1..k``) and the Cypher backends (variable-length paths) override it
    for a single round trip; the conformance suite asserts the override
    returns exactly what the default would. NebulaGraph, TigerGraph and
    PostgreSQL use the default.

    Args:
        vertex_type: Logical type of the anchor vertex.
        key: A complete key of the anchor
            (:mod:`~graflo.architecture.schema.anchor`): the value of a
            one-field primary identity, or a mapping naming exactly the
            primary identity, a derived key's source fields, or one secondary
            identity. A primary key is not looked up: a vertex that does not
            exist has no neighbours. Without *schema*, a bare value is the
            stored id.
        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.
        guard: A trusted predicate, never taken from a request, that every
            vertex and edge on a path must satisfy, the anchor included.
            It names fields as stored, and reserved fields may appear in it.

    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).
        ReservedFieldError: if *filters* name a reserved field.
        AnchorKeyError: if *key* is not a complete key, or a secondary
            identity names several vertices or none.
        NotImplementedError: if a *guard* is given and the backend does not
            declare ``supports_read_guard``.
    """
    guarded = self._guard_kwargs(guard)
    if schema is None:
        if isinstance(key, dict):
            from graflo.architecture.schema.anchor import AnchorNotAKeyError

            raise AnchorNotAKeyError(
                f"an anchor mapping for {vertex_type!r} needs the schema that "
                "says which fields are keys",
                vertex_type=vertex_type,
            )
        return self._graph_neighbors(
            vertex_type,
            key,
            hops=hops,
            direction=direction,
            edge_types=edge_types,
            filters=filters,
            limit=limit,
            schema=None,
            **guarded,
        )
    stored, keys = self._stored_view(schema)
    _refuse_reserved(_reserved_names(schema, stored), filters=filters, key=key)
    anchor = self._anchor_identity(schema, vertex_type, key, guard)
    reached = self._graph_neighbors(
        vertex_type,
        keys.anchor_key(vertex_type, anchor),
        hops=hops,
        direction=direction,
        edge_types=edge_types,
        filters=keys.edge_filter(filters, _walked_edge_ids(schema, edge_types)),
        limit=limit,
        schema=stored,
        **guarded,
    )
    return keys.logical_container(reached, schema.core_schema.edge_config)

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) - relationship_merge_properties: Property names for Cypher MERGE (Neo4j, FalkorDB, Memgraph) so parallel edges differ by weights; on PostgreSQL, the edge table's key columns besides the endpoints - union_fields: List properties written as the union of the stored list and the written one; only a backend that declares supports_union_merge is passed them	{}

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)
            - relationship_merge_properties: Property names for Cypher MERGE
              (Neo4j, FalkorDB, Memgraph) so parallel edges differ by weights;
              on PostgreSQL, the edge table's key columns besides the endpoints
            - union_fields: List properties written as the union of the
              stored list and the written one; only a backend that declares
              ``supports_union_merge`` is passed them
    """

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)

require_read_guard()

Refuse a guarded read on a backend that cannot apply the guard.

Raises:

Type	Description
NotImplementedError	unless the backend declares supports_read_guard.

Source code in graflo/db/conn.py

def require_read_guard(self) -> None:
    """Refuse a guarded read on a backend that cannot apply the guard.

    Raises:
        NotImplementedError: unless the backend declares ``supports_read_guard``.
    """
    if not self.supports_read_guard:
        raise NotImplementedError(
            f"{type(self).__name__} cannot apply a read guard; a guarded read "
            "is refused rather than answered unguarded"
        )

reset_after_use()

End whatever a unit of work left open, before the connection is reused.

A no-op unless the backend holds per-connection state between calls, such as an open transaction.

Source code in graflo/db/conn.py

def reset_after_use(self) -> None:
    """End whatever a unit of work left open, before the connection is reused.

    A no-op unless the backend holds per-connection state between calls,
    such as an open transaction.
    """

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

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
guard	FilterExpression | None	A trusted predicate every match must satisfy, ANDed with the key match; a vertex it hides is never matched.	None

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,
    guard: FilterExpression | None = None,
) -> 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
        guard: A trusted predicate every match must satisfy, ANDed with
            the key match; a vertex it hides is never matched.

    Returns:
        dict: Position in *key_docs* -> every vertex it matched. Positions
            with an unresolvable (partial) key or no match are absent.
    """
    if guard is not None:
        self.require_read_guard()
    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 batched(keys, chunk_size):
        match = build_match_filter(match_keys, chunk)
        # Unchanged without a guard; with one, only a backend that declares
        # guard support gets here, and its fetch_docs parses an expression.
        filters: Any = match if guard is None else conjoin(match, guard)
        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)

sync_begin(schema, edge_types)

Index every vertex identity, and the owner property of edge_types.

Raises:

Type	Description
NotImplementedError	The backend has no sync writer.

Source code in graflo/db/conn.py

def sync_begin(self, schema: Schema, edge_types: list[SyncEdgeType]) -> None:
    """Index every vertex identity, and the owner property of *edge_types*.

    Raises:
        NotImplementedError: The backend has no sync writer.
    """
    raise NotImplementedError(f"{type(self).__name__} has no sync writer")

sync_ensure_vertices(schema, chunk)

Create each vertex whose key is absent, from its row; leave the others.

Source code in graflo/db/conn.py

def sync_ensure_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
    """Create each vertex whose key is absent, from its row; leave the others."""
    raise NotImplementedError(f"{type(self).__name__} has no sync writer")

sync_write_owners(schema, chunk)

Rewrite the chunk's owners whose stored hash differs, as one unit.

Source code in graflo/db/conn.py

def sync_write_owners(self, schema: Schema, chunk: SyncOwnerChunk) -> SyncWritten:
    """Rewrite the chunk's owners whose stored hash differs, as one unit."""
    raise NotImplementedError(f"{type(self).__name__} has no sync writer")

traverse(query, *, schema, guard=None)

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. Names are logical in and out, as for :meth:graph_neighbors; the stored view is resolved once for all seeds.

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
guard	FilterExpression | None	As for :meth:graph_neighbors.	None

Returns:

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

Raises:

Type	Description
ReservedFieldError	if the query names a reserved field.
AnchorKeyError	if a seed's key is not a complete key, or names several vertices or none; seed is its position.
NotImplementedError	if a guard is given and the backend does not declare supports_read_guard.

Source code in graflo/db/conn.py

def traverse(
    self,
    query: Any,
    *,
    schema: Schema,
    guard: FilterExpression | None = None,
) -> 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. Names are logical in and out, as for
    :meth:`graph_neighbors`; the stored view is resolved once for all seeds.

    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.
        guard: As for :meth:`graph_neighbors`.

    Returns:
        GraphContainer: everything reached from any seed, deduplicated.

    Raises:
        ReservedFieldError: if the query names a reserved field.
        AnchorKeyError: if a seed's key is not a complete key, or names
            several vertices or none; ``seed`` is its position.
        NotImplementedError: if a *guard* is given and the backend does not
            declare ``supports_read_guard``.
    """
    guarded = self._guard_kwargs(guard)
    stored, keys = self._stored_view(schema)
    reserved = _reserved_names(schema, stored)
    query.refuse_reserved(reserved)
    for seed in query.seeds:
        _refuse_reserved(reserved, key=seed.get("key"))
    filters = keys.edge_filter(
        query.filters, _walked_edge_ids(schema, query.edge_relations)
    )
    from graflo.architecture.schema.anchor import AnchorKeyError

    # Every seed is resolved before any is walked: one that names no vertex
    # fails the request rather than shrinking its answer.
    anchors: list[tuple[str, str | dict[str, Any]]] = []
    for position, seed in enumerate(query.seeds):
        try:
            anchor = self._anchor_identity(
                schema, seed["vertex_type"], seed["key"], guard
            )
        except AnchorKeyError as error:
            error.seed = position
            raise
        anchors.append((seed["vertex_type"], anchor))
    container = GraphContainer()
    for vertex_type, anchor in anchors:
        reached = self._graph_neighbors(
            vertex_type,
            keys.anchor_key(vertex_type, anchor),
            hops=query.max_hops,
            direction=query.edge_direction,
            edge_types=query.edge_relations,
            filters=filters,
            limit=query.limit,
            schema=stored,
            **guarded,
        )
        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 keys.logical_container(container, schema.core_schema.edge_config)

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
    """

vertex_address(doc, identity_fields)

How this backend addresses doc in an edge query.

Most backends key a vertex on a single value, so the first identity field that is present is the address. NebulaGraph composes a VID from every identity field, so it overrides this — without which a composite identity resolves to a prefix that matches no edge, and the traversal returns empty rather than failing.

Source code in graflo/db/conn.py

def vertex_address(
    self, doc: dict[str, Any], identity_fields: Sequence[str]
) -> str | None:
    """How this backend addresses *doc* in an edge query.

    Most backends key a vertex on a single value, so the first identity
    field that is present is the address. NebulaGraph composes a VID from
    *every* identity field, so it overrides this — without which a
    composite identity resolves to a prefix that matches no edge, and the
    traversal returns empty rather than failing.
    """
    for field in identity_fields:
        value = doc.get(field)
        if value is not None:
            return str(value)
    return None

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"
    SCHEMA_DDL = "supports_schema_ddl"
    INSTANCE_DELETE = "supports_instance_delete"
    READ_GUARD = "supports_read_guard"
    UNION_MERGE = "supports_union_merge"
    SCOPED_DELETE = "supports_scoped_delete"
    MEMBERSHIP_READ = "supports_membership_read"
    BULK_LOAD = "supports_bulk_load"
    SYNC = "supports_sync"

    @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",
            ConnectionCapability.SCHEMA_DDL: "schema migration DDL",
            ConnectionCapability.INSTANCE_DELETE: "deleting vertices and edges",
            ConnectionCapability.READ_GUARD: "a read guard on every element read",
            ConnectionCapability.UNION_MERGE: "union-merged list fields",
            ConnectionCapability.SCOPED_DELETE: "deleting the elements of a scope",
            ConnectionCapability.MEMBERSHIP_READ: "reading a list field by ANY_IN",
            ConnectionCapability.BULK_LOAD: "bulk loading an empty target",
            ConnectionCapability.SYNC: "record sync into a loaded target",
        }[self]

Attributes

BULK_LOAD = 'supports_bulk_load' class-attribute instance-attribute

GRAPH_EXPORT = 'supports_graph_export' class-attribute instance-attribute

GRAPH_READ = 'supports_graph_read' class-attribute instance-attribute

INSTANCE_DELETE = 'supports_instance_delete' class-attribute instance-attribute

MEMBERSHIP_READ = 'supports_membership_read' class-attribute instance-attribute

READ_GUARD = 'supports_read_guard' class-attribute instance-attribute

SCHEMA_DDL = 'supports_schema_ddl' class-attribute instance-attribute

SCHEMA_INTROSPECTION = 'supports_schema_introspection' class-attribute instance-attribute

SCOPED_DELETE = 'supports_scoped_delete' class-attribute instance-attribute

SYNC = 'supports_sync' class-attribute instance-attribute

UNION_MERGE = 'supports_union_merge' class-attribute instance-attribute

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
    relationship_merge_properties: Any
    #: Stored names of the list properties written as a union with what the
    #: edge holds; only a backend that declares ``supports_union_merge`` reads it.
    union_fields: tuple[str, ...] = ()

Attributes

collection_name instance-attribute

dry instance-attribute

relationship_merge_properties instance-attribute

union_fields = () class-attribute instance-attribute

uniq_weight_collections instance-attribute

uniq_weight_fields instance-attribute

Methods:

__init__(dry, collection_name, uniq_weight_fields, uniq_weight_collections, relationship_merge_properties, union_fields=())

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.
    """

Functions:

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),
        relationship_merge_properties=kwargs.pop("relationship_merge_properties", None),
        union_fields=tuple(kwargs.pop("union_fields", ()) or ()),
    )
    if kwargs:
        logger.warning(
            "insert_edges_batch: unsupported keyword arguments ignored: %s",
            sorted(kwargs.keys()),
        )
        kwargs.clear()
    return result

deletable_docs(key_docs, match_keys)

key_docs projected onto match_keys, without those missing a key field.

A document without every key field names no vertex, and must not be read as a pattern that matches many.

Source code in graflo/db/conn.py

def deletable_docs(
    key_docs: list[dict[str, Any]], match_keys: Sequence[str]
) -> list[dict[str, Any]]:
    """*key_docs* projected onto *match_keys*, without those missing a key field.

    A document without every key field names no vertex, and must not be read
    as a pattern that matches many.
    """
    return [
        {key: doc[key] for key in match_keys}
        for doc in key_docs
        if all(doc.get(key) is not None for key in match_keys)
    ]

deletable_endpoints(endpoints, match_keys_source, match_keys_target)

As :func:deletable_docs, for (source, target) pairs of an edge.

Source code in graflo/db/conn.py

def deletable_endpoints(
    endpoints: list[tuple[dict[str, Any], dict[str, Any]]],
    match_keys_source: Sequence[str],
    match_keys_target: Sequence[str],
) -> list[tuple[dict[str, Any], dict[str, Any]]]:
    """As :func:`deletable_docs`, for ``(source, target)`` pairs of an edge."""
    kept: list[tuple[dict[str, Any], dict[str, Any]]] = []
    for source, target in endpoints:
        s = deletable_docs([source], match_keys_source)
        t = deletable_docs([target], match_keys_target)
        if s and t:
            kept.append((s[0], t[0]))
    return kept