graflo.db.arango.conn

ArangoDB connection implementation for graph database operations.

This module implements the Connection interface for ArangoDB, providing specific functionality for graph operations in ArangoDB. It handles:

• Graph and vertex/edge class management (ArangoDB uses collections internally)
• Document and edge operations
• Index creation and management
• AQL query execution
• Batch operations with upsert support

Key Features:

- Graph-based document organization
- Vertex and edge class management
- Persistent, hash, skiplist, and fulltext indexes
- Batch document and edge operations
- AQL query generation and execution

Example

conn = ArangoConnection(config) conn.init_db(schema, recreate_schema=True) conn.upsert_docs_batch(docs, "users", match_keys=["email"])

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

ArangoConnection

Bases: Connection

ArangoDB-specific implementation of the Connection interface.

This class provides ArangoDB-specific implementations for all database operations, including graph management, document operations, and query execution. It uses the ArangoDB Python driver for all operations.

Attributes:

Name	Type	Description
conn		ArangoDB database connection instance
flavor		Database type (ARANGO) for expression flavor mapping

Source code in graflo/db/arango/conn.py

class ArangoConnection(Connection):
    """ArangoDB-specific implementation of the Connection interface.

    This class provides ArangoDB-specific implementations for all database
    operations, including graph management, document operations, and query
    execution. It uses the ArangoDB Python driver for all operations.

    Attributes:
        conn: ArangoDB database connection instance
        flavor: Database type (ARANGO) for expression flavor mapping
    """

    flavor = DBType.ARANGO
    supports_instance_delete = True
    supports_graph_export = True
    # A migration emitter is registered for this backend
    # (graflo/migrate/executor.py); the other six have none.
    supports_schema_ddl = True
    supports_schema_introspection = True
    supports_bulk_load = True

    def __init__(self, config: ArangoConfig):
        """Initialize ArangoDB connection.

        Args:
            config: ArangoDB connection configuration containing URL, credentials,
                and database name
        """
        super().__init__()
        # Store config for later use
        self.config = config
        # Validate required config values
        if config.url is None:
            raise ValueError("ArangoDB connection requires a URL to be configured")
        if config.database is None:
            raise ValueError(
                "ArangoDB connection requires a database name to be configured"
            )

        # ArangoDB accepts empty string for password if None
        password = config.password if config.password is not None else ""
        # ArangoDB has default username "root" if None
        username = config.username if config.username is not None else "root"

        # Store client for system operations
        self.client = ArangoClient(
            hosts=config.url, request_timeout=config.request_timeout
        )
        # Connect to the configured database for regular operations
        self.conn = self.client.db(
            config.database,
            username=username,
            password=password,
        )
        # Store credentials for system operations
        self._username = username
        self._password = password

    def create_database(self, name: str) -> None:
        """Create a new ArangoDB database.

        Database creation/deletion operations must be performed from the _system database.

        Args:
            name: Name of the database to create
        """
        try:
            # Connect to _system database for system operations
            system_db = self.client.db(
                "_system", username=self._username, password=self._password
            )
            if not system_db.has_database(name):
                try:
                    system_db.create_database(name)
                    logger.info(f"Successfully created ArangoDB database '{name}'")
                except Exception as create_error:
                    logger.error(
                        f"Failed to create ArangoDB database '{name}': {create_error}",
                        exc_info=True,
                    )
                    raise
            else:
                logger.debug(f"ArangoDB database '{name}' already exists")
        except Exception as e:
            logger.error(
                f"Error creating ArangoDB database '{name}': {e}",
                exc_info=True,
            )
            raise

    def delete_database(self, name: str) -> None:
        """Delete an ArangoDB database.

        Database creation/deletion operations must be performed from the _system database.

        Args:
            name: Name of the database to delete
        """
        try:
            # Connect to _system database for system operations
            system_db = self.client.db(
                "_system", username=self._username, password=self._password
            )
            if system_db.has_database(name):
                try:
                    system_db.delete_database(name)
                    logger.info(f"Successfully deleted ArangoDB database '{name}'")
                except Exception as delete_error:
                    logger.error(
                        f"Failed to delete ArangoDB database '{name}': {delete_error}",
                        exc_info=True,
                    )
                    raise
            else:
                logger.debug(
                    f"ArangoDB database '{name}' does not exist, skipping deletion"
                )
        except Exception as e:
            logger.error(
                f"Error deleting ArangoDB database '{name}': {e}",
                exc_info=True,
            )
            raise

    def execute(
        self, query: str, bind_vars: dict[str, Any] | None = None, **kwargs: Any
    ) -> Any:
        """Execute an AQL query.

        Args:
            query: AQL query string to execute
            bind_vars: Values the query names as ``@name``
            **kwargs: Additional query parameters

        Returns:
            Cursor: ArangoDB cursor for the query results
        """
        cursor = self.conn.aql.execute(query, bind_vars=bind_vars or None)
        return cursor

    def close(self) -> None:
        """Close the ArangoDB connection."""
        # self.conn.close()

    def _resolve_db_name(self, schema: Schema) -> str:
        db_name = self.config.database
        if not db_name:
            db_name = schema.effective_namespace(DBType.ARANGO)
            self.config.database = db_name
        return db_name

    def _connect_to_database(self, db_name: str) -> None:
        if (
            self.config.database != db_name
            or not hasattr(self, "_db_connected")
            or self._db_connected != db_name
        ):
            self.conn = self.client.db(
                db_name, username=self._username, password=self._password
            )
            self._db_connected = db_name
            logger.debug("Connected to database '%s'", db_name)

    def _database_exists(self, db_name: str) -> bool:
        system_db = self.client.db(
            "_system", username=self._username, password=self._password
        )
        return bool(system_db.has_database(db_name))

    def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
        """Ensure the ArangoDB database exists and connect to it."""
        db_name = self._resolve_db_name(schema)
        try:
            if not self._database_exists(db_name):
                if not create:
                    raise NamespaceNotFoundError(
                        f"ArangoDB database '{db_name}' does not exist. "
                        "Create it manually or call with create_namespace=True."
                    )
                logger.info("Database '%s' does not exist, creating it...", db_name)
                system_db = self.client.db(
                    "_system", username=self._username, password=self._password
                )
                system_db.create_database(db_name)
                logger.info("Successfully created database '%s'", db_name)
            self._connect_to_database(db_name)
        except NamespaceNotFoundError:
            raise
        except Exception as e:
            logger.error("Error ensuring database '%s': %s", db_name, e, exc_info=True)
            raise

    def _schema_has_artifacts(self) -> bool:
        graphs_result = self.conn.graphs()
        collections_result = self.conn.collections()
        has_graphs = isinstance(graphs_result, list) and len(graphs_result) > 0
        non_system: list[str] = []
        if isinstance(collections_result, list):
            for c in collections_result:
                if isinstance(c, dict):
                    name_value = c.get("name")
                    if isinstance(name_value, str) and name_value[0] != "_":
                        non_system.append(name_value)
        return has_graphs or len(non_system) > 0

    def apply_target_schema(
        self,
        schema: Schema,
        *,
        recreate: bool,
        create_namespace: bool = True,
    ) -> None:
        """Define collections, graphs, and indexes in the connected database."""
        self.report_edge_direction_support(schema)
        db_name = self._resolve_db_name(schema)
        try:
            if self._schema_has_artifacts() and not recreate:
                raise SchemaExistsError(
                    f"Schema already exists in database '{db_name}'. "
                    "Set recreate_schema=True to replace, or use clear_data=True "
                    "before ingestion."
                )
            if recreate:
                try:
                    self.delete_graph_structure((), (), delete_all=True)
                    logger.debug("Cleaned database '%s' for fresh start", db_name)
                except Exception as clean_error:
                    logger.warning(
                        "Error during recreate for database '%s': %s",
                        db_name,
                        clean_error,
                        exc_info=True,
                    )
            self.define_schema(schema)
            self.define_indexes(schema)
        except SchemaExistsError:
            raise
        except Exception as e:
            logger.error(
                "Error applying schema to database '%s': %s",
                db_name,
                e,
                exc_info=True,
            )
            raise

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

    def clear_data(self, schema: Schema) -> None:
        """Remove all data from collections without dropping the schema.

        Truncates vertex and edge collections that belong to the schema.
        """
        db_schema = schema.resolve_db_aware(DBType.ARANGO)
        vc = db_schema.vertex_config
        for v in vc.vertex_set:
            cname = vc.vertex_dbname(v)
            if self.conn.has_collection(cname):
                self.conn.collection(cname).truncate()
                logger.debug(f"Truncated vertex collection '{cname}'")
        for edge in db_schema.edge_config.values():
            variants = db_schema.edge_config.runtime(edge).physical_variants()
            for variant in variants:
                cname = cast(str | None, variant.get("storage_name"))
                if cname and self.conn.has_collection(cname):
                    self.conn.collection(cname).truncate()
                    logger.debug(f"Truncated edge collection '{cname}'")

    def define_schema(self, schema: Schema, *, identity_indexes: bool = True) -> None:
        """Define ArangoDB collections based on schema.

        Args:
            schema: Schema containing collection definitions
            identity_indexes: Index each new vertex collection on its identity.
        """
        assert_schema_supported(self.flavor, schema)
        self.define_vertex_classes(schema, identity_indexes=identity_indexes)
        self.define_edge_classes(
            list(schema.core_schema.edge_config.values()), schema=schema
        )

    def define_vertex_classes(
        self, schema: Schema, *, identity_indexes: bool = True
    ) -> None:
        """Define vertex collections in ArangoDB.

        Creates vertex collections for both connected and disconnected vertices,
        organizing them into appropriate graphs.

        Args:
            schema: Schema containing vertex definitions
            identity_indexes: Index each new collection on its vertex identity.
        """
        db_schema = schema.resolve_db_aware(DBType.ARANGO)
        vertex_config = db_schema.vertex_config

        def index(vertex: str) -> Index | None:
            return vertex_config.index(vertex) if identity_indexes else None

        disconnected_vertex_collections = (
            set(vertex_config.vertex_set) - db_schema.edge_config.vertices
        )
        for item in db_schema.edge_config.values():
            u, v = item.source, item.target
            gname = db_schema.edge_config.runtime(item).graph_name()
            if not gname:
                logger.warning(
                    f"Edge {item.source} -> {item.target} has no graph_name, skipping"
                )
                continue
            logger.info(f"{item.source}, {item.target}, {gname}")
            if self.conn.has_graph(gname):
                g_result = self.conn.graph(gname)
            else:
                g_result = self.conn.create_graph(gname)

            # Type narrowing: ensure g is a Graph instance
            g: Graph | None = None
            if isinstance(g_result, Graph):
                g = g_result
            elif g_result is not None:
                # If it's not a Graph, log warning and skip
                logger.warning(f"Graph {gname} is not a Graph instance, skipping")
                continue

            _ = self.create_collection(vertex_config.vertex_dbname(u), index(u), g)
            _ = self.create_collection(vertex_config.vertex_dbname(v), index(v), g)
        for v in disconnected_vertex_collections:
            _ = self.create_collection(vertex_config.vertex_dbname(v), index(v), None)

    def define_edge_classes(
        self, edges: list[Edge], schema: Schema | None = None
    ) -> None:
        """Define edge classes in ArangoDB.

        Creates edge collections and their definitions in the appropriate graphs.

        Args:
            edges: List of edge configurations to create
            schema: Optional schema used to expand purpose-based physical variants
        """
        db_schema = (
            schema.resolve_db_aware(DBType.ARANGO) if schema is not None else None
        )
        for item in edges:
            variants = (
                db_schema.edge_config.runtime(item).physical_variants()
                if db_schema is not None
                else [
                    {
                        "purpose": None,
                        "storage_name": f"{item.source}_{item.target}_edges",
                        "graph_name": f"{item.source}_{item.target}_edges",
                        "indexes": [],
                    }
                ]
            )
            for variant in variants:
                gname = cast(str | None, variant.get("graph_name"))
                if not gname:
                    logger.warning("Edge has no graph_name, skipping")
                    continue
                if self.conn.has_graph(gname):
                    g_result = self.conn.graph(gname)
                else:
                    g_result = self.conn.create_graph(gname)
                # Type guard: ensure g is a Graph instance
                if not isinstance(g_result, Graph):
                    logger.warning(f"Graph {gname} is not a Graph instance, skipping")
                    continue
                g = g_result
                collection_name = cast(str | None, variant.get("storage_name"))
                if not collection_name:
                    logger.warning("Edge has no database_name, skipping")
                    continue
                if not g.has_edge_definition(collection_name):
                    if db_schema is None:
                        logger.warning(
                            "DB-aware schema is required, skipping edge definition"
                        )
                        continue
                    runtime = db_schema.edge_config.runtime(item)
                    _ = g.create_edge_definition(
                        edge_collection=collection_name,
                        from_vertex_collections=[runtime.source_storage],
                        to_vertex_collections=[runtime.target_storage],
                    )

    def _add_index(self, general_collection: Any, index: Index) -> Any | None:
        """Add an index to an ArangoDB collection.

        Supports persistent, hash, skiplist, and fulltext indexes.

        Args:
            general_collection: ArangoDB collection to add index to
            index: Index configuration to create

        Returns:
            IndexHandle: Handle to the created index, or None if index type is not supported
        """
        data = index.db_form(DBType.ARANGO)
        ih: Any | None = None
        if index.type == IndexType.PERSISTENT or index.type == IndexType.HASH:
            ih = general_collection.add_index(data)
        elif index.type == IndexType.SKIPLIST:
            ih = general_collection.add_skiplist_index(
                fields=index.fields, unique=index.unique
            )
        elif index.type == IndexType.FULLTEXT:
            ih = general_collection.add_index(
                data={"fields": index.fields, "type": "fulltext"}
            )
        return ih

    def define_vertex_indexes(
        self, vertex_config: VertexConfig, schema: Schema | None = None
    ) -> None:
        """Define indexes for vertex collections.

        Creates indexes for each vertex collection based on the configuration.
        Identity index is created at collection creation; this adds secondary indexes only.

        Args:
            vertex_config: Vertex configuration containing index definitions
        """
        db_vertex = (
            schema.resolve_db_aware(DBType.ARANGO).vertex_config if schema else None
        )
        for c in vertex_config.vertex_set:
            collection_name = db_vertex.vertex_dbname(c) if db_vertex else c
            general_collection = self.conn.collection(collection_name)
            ixs = general_collection.indexes()
            field_combinations: list[tuple[Any, ...]] = []
            if isinstance(ixs, list):
                for ix in ixs:
                    if isinstance(ix, dict):
                        fields_value = ix.get("fields")
                        if isinstance(fields_value, (list, tuple)):
                            field_combinations.append(tuple(fields_value))
            index_list = (
                schema.db_profile.vertex_secondary_indexes(c)
                if schema is not None
                else []
            )
            for index_obj in index_list:
                if tuple(index_obj.fields) not in field_combinations:
                    self._add_index(general_collection, index_obj)

    def define_edge_indexes(
        self, edges: list[Edge], schema: Schema | None = None
    ) -> None:
        """Define indexes for edge collections.

        Creates indexes for each edge collection based on the configuration.

        Args:
            edges: List of edge configurations containing index definitions
        """
        if schema is None:
            logger.warning("Schema required for edge index naming, skipping")
            return
        db_schema = schema.resolve_db_aware(DBType.ARANGO)
        for edge in edges:
            variants = db_schema.edge_config.runtime(edge).physical_variants()
            for variant in variants:
                collection_name = cast(str | None, variant.get("storage_name"))
                if not collection_name:
                    logger.warning("Edge has no database_name, skipping index creation")
                    continue
                general_collection = self.conn.collection(collection_name)
                index_list = cast(list[Index], variant.get("indexes", []))
                for index_obj in index_list:
                    self._add_index(general_collection, index_obj)

    def fetch_indexes(self, db_class_name: str | None = None) -> dict[str, Any]:
        """Fetch all indexes from the database.

        Args:
            db_class_name: Optional collection name to fetch indexes for

        Returns:
            dict: Mapping of collection names to their indexes
        """
        classes: list[Any] = []
        if db_class_name is None:
            classes_result = self.conn.collections()
            if isinstance(classes_result, list):
                classes = classes_result
        elif self.conn.has_collection(db_class_name):
            classes = [self.conn.collection(db_class_name)]

        r: dict[str, Any] = {}
        for cname in classes:
            if isinstance(cname, dict):
                cname_dict = cast(dict[str, Any], cname)
                name_value = cname_dict.get("name")
                if isinstance(name_value, str):
                    c = self.conn.collection(name_value)
                    r[name_value] = c.indexes()
        return r

    def create_collection(
        self,
        db_class_name: str,
        index: None | Index = None,
        g: Graph | None = None,
    ) -> Any | None:
        """Create a new vertex or edge class (ArangoDB uses collections internally).

        Args:
            db_class_name: Name of the vertex/edge class to create (ArangoDB collection name)
            index: Optional index to create on the class
            g: Optional graph to create the class in

        Returns:
            IndexHandle: Handle to the created index if one was created, None otherwise
        """
        if not self.conn.has_collection(db_class_name):
            if g is not None:
                _ = g.create_vertex_collection(db_class_name)
            else:
                self.conn.create_collection(db_class_name)
            general_collection = self.conn.collection(db_class_name)
            if index is not None and index.fields != ["_key"]:
                ih = self._add_index(general_collection, index)
                return ih
            else:
                return None

    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 (vertex/edge classes and graphs) from ArangoDB.

        In ArangoDB:
        - Collections (internal): Container for vertices (vertex collections) and edges (edge collections)
        - Graphs: Named graphs that connect vertex and edge collections

        Args:
            vertex_types: Vertex/edge class names to delete (ArangoDB collection names)
            graph_names: Graph names to delete
            delete_all: If True, delete all non-system vertex/edge classes and graphs
        """

        def _graph_name(graph_item: Any) -> str | None:
            if not isinstance(graph_item, dict):
                return None
            graph_dict = cast(dict[str, Any], graph_item)
            for key in ("name", "_key"):
                value = graph_dict.get(key)
                if isinstance(value, str) and value:
                    return value
            return None

        cnames: list[str] = list(vertex_types)
        gnames: list[str] = list(graph_names)
        logger.info("vertex/edge classes (non system, ArangoDB collections):")
        collections_result = self.conn.collections()
        if isinstance(collections_result, list):
            filtered_collections: list[dict[str, Any]] = []
            for c in collections_result:
                if isinstance(c, dict):
                    name_value = c.get("name")
                    if isinstance(name_value, str) and name_value[0] != "_":
                        filtered_collections.append(c)
            logger.info(filtered_collections)
        else:
            logger.info([])

        if delete_all:
            collections_result = self.conn.collections()
            graphs_result = self.conn.graphs()
            cnames = []
            if isinstance(collections_result, list):
                for c in collections_result:
                    if isinstance(c, dict):
                        name_value = c.get("name")
                        if isinstance(name_value, str) and name_value[0] != "_":
                            cnames.append(name_value)
            gnames = []
            if isinstance(graphs_result, list):
                for g in graphs_result:
                    name_value = _graph_name(g)
                    if name_value is not None:
                        gnames.append(name_value)

        # Delete graphs first. Retry because graph metadata can be stale after deletes.
        for _ in range(3):
            deleted_any = False
            for gn in gnames:
                if self.conn.has_graph(gn):
                    self.conn.delete_graph(gn)
                    deleted_any = True
            if not deleted_any:
                break
            # Refresh remaining graph names before next pass.
            refreshed_graphs = self.conn.graphs()
            gnames = []
            if isinstance(refreshed_graphs, list):
                for g in refreshed_graphs:
                    name_value = _graph_name(g)
                    if name_value is not None:
                        gnames.append(name_value)

        logger.info("graphs (after delete operation):")
        logger.info(self.conn.graphs())

        # Delete collections. If any collection is still attached to a graph,
        # remove graphs again and retry once.
        for attempt in range(2):
            blocked = False
            for cn in cnames:
                if not self.conn.has_collection(cn):
                    continue
                try:
                    self.conn.delete_collection(cn)
                except Exception as e:
                    if "ERR 1942" in str(e) and attempt == 0:
                        blocked = True
                        continue
                    raise
            if not blocked:
                break
            refreshed_graphs = self.conn.graphs()
            if isinstance(refreshed_graphs, list):
                for g in refreshed_graphs:
                    gn = _graph_name(g)
                    if gn is not None and self.conn.has_graph(gn):
                        self.conn.delete_graph(gn)

        logger.info(
            "vertex/edge classes (after delete operation, ArangoDB collections):"
        )
        collections_result = self.conn.collections()
        if isinstance(collections_result, list):
            collection_names: list[str] = []
            for c in collections_result:
                if isinstance(c, dict):
                    name_value = c.get("name")
                    if isinstance(name_value, str) and name_value[0] != "_":
                        collection_names.append(name_value)
            logger.info(collection_names)
        else:
            logger.info([])

        logger.info("graphs:")
        logger.info(self.conn.graphs())

    def get_collections(self) -> list[dict[str, Any]]:
        """Get all vertex and edge classes in the database (ArangoDB collections).

        Returns:
            list: List of class information dictionaries (ArangoDB collection info)
        """
        result = self.conn.collections()
        if isinstance(result, list):
            return [item if isinstance(item, dict) else {} for item in result]
        return []

    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 using AQL.

        Documents are matched on *match_keys*; each gets the ``_key`` those
        values determine (:func:`graflo.db.arango.keys.vertex_key`), so edges
        can name it without a lookup.

        Args:
            docs: List of documents to upsert
            class_name: Collection name to upsert into
            match_keys: Keys to match for upsert operation
            **kwargs: Additional options:
                - dry: If True, don't execute the query
                - update_keys: Keys to update on match
                - filter_uniques: If True, filter duplicate documents
        """
        dry = kwargs.pop("dry", False)
        update_keys = kwargs.pop("update_keys", None)
        filter_uniques = kwargs.pop("filter_uniques", True)

        if not docs:
            return
        if filter_uniques:
            docs = pick_unique_dict(docs)
        docs = vertex_documents(docs, match_keys, strict=False)
        docs_json = json.dumps(docs, default=json_serializer)
        if not match_keys:
            upsert_clause = ""
            update_clause = ""
        else:
            upsert_clause = ", ".join([f'"{k}": doc.{k}' for k in match_keys])
            upsert_clause = f"UPSERT {{{upsert_clause}}}"

            if isinstance(update_keys, list):
                update_clause = ", ".join([f'"{k}": doc.{k}' for k in update_keys])
                update_clause = f"{{{update_clause}}}"
            elif update_keys == "doc":
                update_clause = "doc"
            else:
                update_clause = "{}"
            update_clause = f"UPDATE {update_clause}"

        options = "OPTIONS {ignoreErrors: true}"

        q_update = f"""FOR doc in {docs_json}
                            {upsert_clause}
                            INSERT doc
                            {update_clause}
                                IN {class_name} {options}"""
        if not dry:
            self.execute(q_update)

    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, ...] = ("_key",),
        match_keys_target: tuple[str, ...] = ("_key",),
        filter_uniques: bool = True,
        head: int | None = None,
        **kwargs: Any,
    ) -> None:
        """Upsert a batch of edges on their ``_key``.

        An edge is its endpoints, relation and identity fields, digested into
        its ``_key``; a row with the same identity updates the edge's other
        properties. ``_from`` and ``_to`` are built from the endpoints' match
        key values, so no vertex is looked up and an absent endpoint is not
        checked. Rows missing an endpoint value are dropped.

        Args:
            docs_edges: ``[source_doc, target_doc, props]`` rows
            source_class: Source vertex collection
            target_class: Target vertex collection
            relation_name: Relation stored on each edge, if any
            match_keys_source: The source's identity fields, in identity order
            match_keys_target: The target's identity fields, in identity order
            filter_uniques: If True, filter duplicate rows
            head: Optional limit on number of edges to insert
            **kwargs: Additional options:
                - dry: If True, don't execute the query
                - collection_name: Edge collection name (defaults to {source_class}_{target_class}_edges if not provided)
                - uniq_weight_fields: The edge's identity fields (parallel edges)
                - uniq_weight_collections: Extra identity fields (legacy; list of names)
                - relationship_merge_properties: Ignored (Cypher backends only)
        """
        opts = consume_insert_edges_kwargs(kwargs)
        collection_name = opts.collection_name or f"{source_class}_{target_class}_edges"

        if not isinstance(docs_edges, list):
            return
        if head is not None:
            docs_edges = docs_edges[:head]
        if filter_uniques:
            docs_edges = pick_unique_dict(docs_edges)
        docs = edge_documents(
            docs_edges,
            source_class,
            target_class,
            match_keys_source,
            match_keys_target,
            relation_name,
            _arango_edge_uniq_weight_keys(
                opts.uniq_weight_fields, opts.uniq_weight_collections
            ),
            strict=False,
            fuse=True,
        )
        if not docs or opts.dry:
            return
        self.execute(
            arango_upsert_edges_query(
                collection_name, json.dumps(docs, default=json_serializer)
            )
        )

    # -- bulk load ---------------------------------------------------------------

    def bulk_begin(
        self,
        schema: Schema,
        cfg: BulkConfig,
        *,
        recreate: bool = False,
        resume: bool = False,
    ) -> BulkPlan:
        """Guard the first write and create the collections, without their indexes.

        Identity and secondary indexes are built by :meth:`bulk_finish`; the
        primary index on ``_key`` is all an import needs.

        Raises:
            SchemaExistsError: The database holds collections or graphs and
                neither *recreate* nor *resume* is set.
        """
        assert_schema_supported(self.flavor, schema)
        self.report_edge_direction_support(schema)
        self.ensure_target_namespace(schema, create=True)
        if not resume:
            if recreate:
                self.delete_graph_structure((), (), delete_all=True)
            elif self._schema_has_artifacts():
                raise SchemaExistsError(
                    f"Schema already exists in database "
                    f"'{self._resolve_db_name(schema)}'. Set recreate_schema=True "
                    "to replace it."
                )
        self.define_schema(schema, identity_indexes=False)
        return BulkPlan()

    def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
        """Import the chunk's documents, each under the ``_key`` its identity gives."""
        docs = vertex_documents(chunk.rows, chunk.match_keys, strict=True)
        self._bulk_import(chunk.class_name, docs, replay=chunk.replay)

    def bulk_prepare_edges(self, schema: Schema) -> None:
        """Nothing: edges carry ``_from`` and ``_to`` built from endpoint keys."""

    def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
        """Import the chunk's edges, each under the ``_key`` its identity gives."""
        docs = edge_documents(
            chunk.rows,
            chunk.source_class,
            chunk.target_class,
            chunk.match_keys_source,
            chunk.match_keys_target,
            chunk.relation_name,
            chunk.merge_properties,
            strict=True,
            fuse=False,
        )
        collection = (
            chunk.collection_name or f"{chunk.source_class}_{chunk.target_class}_edges"
        )
        self._bulk_import(collection, docs, replay=chunk.replay)

    def bulk_finish(self, schema: Schema) -> BulkCounts:
        """Build the identity and secondary indexes; count what the database holds."""
        vc = schema.resolve_db_aware(DBType.ARANGO).vertex_config
        for vertex in vc.vertex_set:
            index = vc.index(vertex)
            if index.fields and index.fields != ["_key"]:
                # Idempotent: ArangoDB returns an index that already exists.
                self._add_index(self.conn.collection(vc.vertex_dbname(vertex)), index)
        self.define_indexes(schema)
        vertices = {
            vc.vertex_dbname(vertex): int(
                cast(int, self.conn.collection(vc.vertex_dbname(vertex)).count())
            )
            for vertex in vc.vertex_set
        }
        edges: dict[StoredEdgeId, int] = {}
        for collection, stored in edge_collections(schema).items():
            if len(stored) == 1 and stored[0][1] is not None:
                count = cast(int, self.conn.collection(collection).count())
                edges[stored[0]] = edges.get(stored[0], 0) + int(count)
                continue
            # Several edges, or a relation set per row, share the collection.
            query = edge_count_query(_arango_safe_collection_name(collection))
            for source, relation, target, n in get_data_from_cursor(
                self.execute(query)
            ):
                key = (source, relation, target)
                edges[key] = edges.get(key, 0) + int(n)
        return BulkCounts(vertices=vertices, edges=edges)

    def _bulk_import(
        self, collection: str, docs: list[dict[str, Any]], *, replay: bool
    ) -> None:
        """POST *docs* to ``/_api/import`` as JSON Lines, a few MiB per request.

        Raises:
            ArangoImportError: A request did not store every document it sent.
        """
        for body, sent in import_bodies(docs):
            request = Request(
                method="post",
                endpoint="/_api/import",
                params=import_params(collection),
                data=body,
                write=collection,
            )
            response = self.conn.conn.send_request(request)
            if not response.is_success:
                raise ArangoImportError(
                    f"import into {collection}: HTTP {response.status_code} "
                    f"{response.error_message or response.body}"
                )
            check_import(response.body, sent=sent, replay=replay, collection=collection)

    def insert_return_batch(self, docs: list[dict[str, Any]], class_name: str) -> str:
        """Insert documents and return the AQL query string.

        Note: ArangoDB-specific behavior - returns query string instead of executing.
        This allows for deferred execution and is used by caster.py and tests.

        Args:
            docs: Documents to insert
            class_name: Collection to insert into

        Returns:
            str: AQL query string for the operation (can be executed with execute())
        """
        docs_str = json.dumps(docs, default=json_serializer)
        query0 = f"""FOR doc in {docs_str}
              INSERT doc
              INTO {class_name}
              LET inserted = NEW
              RETURN {{_key: inserted._key}}
        """
        return query0

    def _edge_collection_names(self) -> list[str]:
        """Every non-system edge collection of the database."""
        names: list[str] = []
        collections = self.conn.collections()
        if not isinstance(collections, list):
            return names
        for collection in collections:
            if not isinstance(collection, dict):
                continue
            name = collection.get("name")
            if (
                isinstance(name, str)
                and not name.startswith("_")
                and _arango_collection_kind(collection.get("type")) == "edge"
            ):
                names.append(name)
        return names

    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 documents and every edge, in any edge collection, that touches them."""
        rows = deletable_docs(key_docs, match_keys)
        if not rows:
            return
        edge_collections = self._edge_collection_names()
        collection = _arango_safe_collection_name(class_name)
        for start in range(0, len(rows), chunk_size):
            removed = list(
                get_data_from_cursor(
                    self.conn.aql.execute(
                        arango_remove_documents_query(collection, match_keys),
                        bind_vars={"rows": rows[start : start + chunk_size]},
                    )
                )
            )
            if not removed:
                continue
            for edges in edge_collections:
                self.conn.aql.execute(
                    arango_remove_incident_edges_query(
                        _arango_safe_collection_name(edges)
                    ),
                    bind_vars={"ids": removed},
                )

    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 edges between endpoint pairs, located by their keys."""
        pairs = deletable_endpoints(endpoints, match_keys_source, match_keys_target)
        if not pairs:
            return
        query = arango_remove_edges_query(
            _arango_safe_collection_name(
                collection_name or f"{source_class}_{target_class}_edges"
            ),
            _arango_safe_collection_name(source_class),
            _arango_safe_collection_name(target_class),
            match_keys_source,
            match_keys_target,
            with_relation=relation_name is not None,
        )
        rows = [{"s": source, "t": target} for source, target in pairs]
        for start in range(0, len(rows), chunk_size):
            bind_vars: dict[str, Any] = {"rows": rows[start : start + chunk_size]}
            if relation_name is not None:
                bind_vars["relation"] = relation_name
            self.conn.aql.execute(query, bind_vars=bind_vars)

    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: None | FilterExpression | list[Any] | dict[str, Any] = 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: Collection to check in
            match_keys: Keys to match documents
            keep_keys: Keys to keep in result
            flatten: If True, flatten the result into a list
            filters: Additional query filters

        Returns:
            list | dict: Documents that exist in the database, either as a
                flat list or a dictionary mapping batch indices to documents
        """
        q0, bind_vars = fetch_fields_query(
            collection_name=class_name,
            docs=batch,
            match_keys=match_keys,
            keep_keys=keep_keys,
            filters=filters,
        )
        # {"__i": i, "_group": [doc]}
        cursor = self.execute(q0, bind_vars=bind_vars)

        if flatten:
            rdata = []
            for item in get_data_from_cursor(cursor):
                group = item.pop("_group", [])
                rdata += [sub_item for sub_item in group]
            return rdata
        else:
            rdata_dict: dict[int, list[dict[str, Any]]] = {}
            for item in get_data_from_cursor(cursor):
                __i = item.pop("__i")
                group = item.pop("_group", [])
                rdata_dict[__i] = group
            return rdata_dict

    def fetch_docs(
        self,
        class_name: str,
        filters: None | FilterExpression | list[Any] | dict[str, Any] = 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 collection.

        Args:
            class_name: Collection to fetch from
            filters: Query filters
            limit: Maximum number of documents to return
            return_keys: Keys to return
            unset_keys: Keys to unset

        Returns:
            list: Fetched documents
        """
        params = BoundParams(ExpressionFlavor.AQL)
        filter_clause = render_filters(filters, doc_name="d", params=params)

        if return_keys is None:
            if unset_keys is None:
                return_clause = "d"
            else:
                tmp_clause = ", ".join([f'"{item}"' for item in unset_keys])
                return_clause = f"UNSET(d, {tmp_clause})"
        else:
            if unset_keys is None:
                tmp_clause = ", ".join([f'"{item}"' for item in return_keys])
                return_clause = f"KEEP(d, {tmp_clause})"
            else:
                raise ValueError("both return_keys and unset_keys are set")

        if limit is not None and isinstance(limit, int):
            limit_clause = f"LIMIT {limit}"
        else:
            limit_clause = ""

        q = (
            f"FOR d in {class_name}"
            f"  {filter_clause}"
            f"  {limit_clause}"
            f"  RETURN {return_clause}"
        )
        cursor = self.execute(q, bind_vars=params.values)
        return get_data_from_cursor(cursor)

    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] | FilterExpression | 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 from ArangoDB using AQL.

        Reverse lookup is free here: an edge collection's edge index covers
        ``_from`` *and* ``_to``, so ``IN`` and ``ANY`` cost the same as ``OUT``.

        Args:
            from_type: Anchor vertex collection name
            from_id: Anchor vertex ID (can be _key or _id)
            edge_type: Optional edge collection name to filter by
            to_type: Optional collection of the other endpoint
            to_id: Optional 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
            **kwargs: Additional parameters

        Returns:
            list: List of fetched edges
        """
        # Convert from_id to _id format if needed
        if not from_id.startswith(from_type):
            # Assume it's a _key, convert to _id
            from_vertex_id = f"{from_type}/{from_id}"
        else:
            from_vertex_id = from_id

        # Build AQL query to fetch edges
        # Start with basic edge traversal
        if edge_type:
            edge_collection = edge_type
        else:
            # If no edge_type specified, we need to search all edge collections
            # This is a simplified version - in practice you might want to list all edge collections
            raise ValueError("edge_type is required for ArangoDB edge fetching")

        params = BoundParams(ExpressionFlavor.AQL)
        # Already a `FILTER ...` clause, or empty.
        filter_clause = render_filters(filters, doc_name="e", params=params)

        to_vertex_id: str | None = None
        if to_id and to_type:
            to_vertex_id = to_id if to_id.startswith(to_type) else f"{to_type}/{to_id}"

        bind_vars: dict[str, Any] = {}
        anchor_clause = _arango_edge_anchor_clause(
            from_vertex_id, direction, to_type, to_vertex_id, bind_vars
        )
        bind_vars.update(params.values)

        query = f"""
            FOR e IN {edge_collection}
                FILTER {anchor_clause}
                {filter_clause}
                {f"LIMIT {limit}" if limit else ""}
                RETURN e
        """

        cursor = self.execute(query, bind_vars=bind_vars)
        result = list(get_data_from_cursor(cursor))

        # Apply projection
        if return_keys is not None:
            result = [
                {k: doc.get(k) for k in return_keys if k in doc} for doc in result
            ]
        elif unset_keys is not None:
            result = [
                {k: v for k, v in doc.items() if k not in unset_keys} for doc in result
            ]

        return result

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

        Args:
            class_name: Collection to aggregate
            aggregation_function: Type of aggregation to perform
            discriminant: Field to group by
            aggregated_field: Field to aggregate
            filters: Query filters

        Returns:
            list: Aggregation results
        """
        params = BoundParams(ExpressionFlavor.AQL)
        filter_clause = render_filters(filters, doc_name="doc", params=params)

        if (
            aggregated_field is not None
            and aggregation_function != AggregationType.COUNT
        ):
            group_unit = f"g[*].doc.{aggregated_field}"
        else:
            group_unit = "g"

        if discriminant is not None:
            collect_clause = f"COLLECT value = doc['{discriminant}'] INTO g"
            return_clause = f"""{{ '{discriminant}' : value, '_value': {aggregation_function}({group_unit})}}"""
        else:
            if (
                aggregated_field is None
                and aggregation_function == AggregationType.COUNT
            ):
                collect_clause = (
                    f"COLLECT AGGREGATE value =  {aggregation_function} (doc)"
                )
            else:
                collect_clause = (
                    "COLLECT AGGREGATE value ="
                    f" {aggregation_function}(doc['{aggregated_field}'])"
                )
            return_clause = """{ '_value' : value }"""

        q = f"""FOR doc IN {class_name}
                    {filter_clause}
                    {collect_clause}
                    RETURN {return_clause}"""

        cursor = self.execute(q, bind_vars=params.values)
        data = get_data_from_cursor(cursor)
        return data

    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: None | FilterExpression | list[Any] | dict[str, Any] = None,
    ) -> list[dict[str, Any]]:
        """Keep documents that don't exist in the database.

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

        Returns:
            list: Documents that don't exist in the database
        """
        present_docs_keys = self.fetch_present_documents(
            batch=batch,
            class_name=class_name,
            match_keys=match_keys,
            keep_keys=keep_keys,
            flatten=False,
            filters=filters,
        )

        if not isinstance(present_docs_keys, dict):
            raise TypeError(f"expected dict, got {type(present_docs_keys).__name__}")

        if any(len(v) > 1 for v in present_docs_keys.values()):
            logger.warning(
                "fetch_present_documents returned multiple docs per filtering condition"
            )

        absent_indices = sorted(set(range(len(batch))) - set(present_docs_keys.keys()))
        batch_absent = [batch[j] for j in absent_indices]
        return batch_absent

    def update_to_numeric(self, collection_name: str, field: str) -> str:
        """Update a field to numeric type in all documents.

        Args:
            collection_name: Vertex/edge class name to update (ArangoDB collection name)
            field: Field to convert to numeric

        Returns:
            str: AQL query string for the operation
        """
        s1 = f"FOR p IN {collection_name} FILTER p.{field} update p with {{"
        s2 = f"{field}: TO_NUMBER(p.{field}) "
        s3 = f"}} in {collection_name}"
        q0 = s1 + s2 + s3
        return q0

    def introspect_graph_schema(
        self,
        schema_name: str | None = None,
        *,
        sample_limit: int = 100,
    ) -> Schema:
        """Infer a graflo Schema from ArangoDB via collection sampling."""
        resolved_name = schema_name or self.config.database or "arango"
        collections = self.get_collections()
        vertex_collections: list[str] = []
        edge_collections: list[str] = []
        for coll in collections:
            name = coll.get("name")
            if not name or name.startswith("_"):
                continue
            kind = _arango_collection_kind(coll.get("type"))
            if kind == "edge":
                edge_collections.append(name)
            elif kind == "document":
                vertex_collections.append(name)

        vertices: list[GraphVertexIntrospection] = []
        for coll_name in vertex_collections:
            safe = _arango_safe_collection_name(coll_name)
            props_query = f"""
                FOR doc IN {safe}
                    LIMIT {sample_limit}
                    FOR key IN ATTRIBUTES(doc)
                        FILTER key NOT IN ['_id', '_key', '_rev']
                        RETURN DISTINCT key
            """
            prop_rows = get_data_from_cursor(self.execute(props_query))
            properties = [row for row in prop_rows if isinstance(row, str)]
            vertices.append(
                GraphVertexIntrospection(
                    name=coll_name,
                    properties=properties,
                    identity=infer_identity_fields(properties),
                )
            )

        edges: list[GraphEdgeIntrospection] = []
        for coll_name in edge_collections:
            safe = _arango_safe_collection_name(coll_name)
            pattern_query = f"""
                FOR edge IN {safe}
                    LIMIT {sample_limit}
                    COLLECT source = SPLIT(edge._from, '/')[0],
                            target = SPLIT(edge._to, '/')[0]
                    RETURN {{ source, target }}
            """
            pattern_rows = get_data_from_cursor(self.execute(pattern_query))
            edge_props_query = f"""
                FOR edge IN {safe}
                    LIMIT {sample_limit}
                    FOR key IN ATTRIBUTES(edge)
                        FILTER key NOT IN ['_id', '_key', '_rev', '_from', '_to']
                        RETURN DISTINCT key
            """
            edge_prop_rows = get_data_from_cursor(self.execute(edge_props_query))
            edge_properties = [row for row in edge_prop_rows if isinstance(row, str)]
            relation: str | None = None
            if "relation" in edge_properties:
                relation = "relation"
            seen_pairs: set[tuple[str, str]] = set()
            for row in pattern_rows:
                if not isinstance(row, dict):
                    continue
                source = row.get("source")
                target = row.get("target")
                if not source or not target:
                    continue
                pair = (source, target)
                if pair in seen_pairs:
                    continue
                seen_pairs.add(pair)
                edges.append(
                    GraphEdgeIntrospection(
                        source=source,
                        target=target,
                        relation=relation,
                        properties=edge_properties,
                        collection_name=coll_name,
                    )
                )

        introspection = GraphIntrospectionResult(
            name=resolved_name,
            vertices=vertices,
            edges=edges,
        )
        return GraphSchemaInferencer(db_flavor=DBType.ARANGO).infer_schema(
            introspection, schema_name=resolved_name
        )

    def _graph_neighbors(
        self,
        vertex_type: str,
        key: str | dict[str, Any],
        *,
        hops: int = 1,
        direction: EdgeDirection = EdgeDirection.OUT,
        edge_types: Sequence[str] | None = None,
        filters: Any | None = None,
        limit: int | None = None,
        schema: Schema | None = None,
    ) -> GraphContainer:
        """Bounded neighbourhood via AQL ``FOR v, e IN 1..k``.

        Native rather than composed from :meth:`fetch_edges`, because that method
        anchors on a document ``_id`` (``collection/key``) while callers hold a
        *logical* identity. Resolving the anchor inside the traversal query keeps
        Arango's internal key out of the agent-facing contract entirely.

        Returns the reached vertices and the edges of the paths that reached
        them, shaped as the default walk returns them: edge rows by edge id,
        their ends under ``source`` and ``target``, at most *limit* of each.
        """
        from graflo.architecture.graph_types import EdgeId
        from graflo.architecture.schema.edge import Edge
        from graflo.db.traversal import (
            _row_marker,
            _vertex_identity_value,
            check_anchor_fields,
            edge_query_name,
        )

        if hops < 1:
            raise ValueError(f"hops must be >= 1, got {hops}")
        if schema is None:
            raise ValueError(
                "graph_neighbors requires a schema: logical vertex and relation "
                "names cannot be resolved to storage names without one"
            )
        if vertex_type not in schema.core_schema.vertex_config.vertex_set:
            raise ValueError(
                f"Unknown vertex type {vertex_type!r}; declared: "
                f"{sorted(schema.core_schema.vertex_config.vertex_set)}"
            )

        db_aware = schema.resolve_db_aware(DBType.ARANGO)
        anchor_collection = _arango_safe_collection_name(
            db_aware.vertex_config.vertex_dbname(vertex_type)
        )
        identity_fields = db_aware.vertex_config.identity_fields(vertex_type)
        # The anchor's field names and values travel as bind parameters, and a
        # field name must be one the schema declares: both arrive with the
        # request, so neither may become query text.
        if isinstance(key, dict):
            anchor_fields = list(key.items())
            check_anchor_fields(schema, db_aware, vertex_type, key)
        else:
            anchor_fields = [(identity_fields[0] if identity_fields else "_key", key)]
        bind_vars: dict[str, Any] = {}
        clauses: list[str] = []
        for position, (field, value) in enumerate(anchor_fields):
            bind_vars[f"field{position}"] = field
            bind_vars[f"value{position}"] = value
            clauses.append(f"doc.@field{position} == @value{position}")
        anchor_filter = " AND ".join(clauses)

        # Every allowed relation, not only those touching the anchor type: a
        # walk of more than one hop crosses relations (work_order -> machine <-
        # line <- product), and a traversal restricted to the anchor's own edge
        # collections stops at the first hop.
        allowed = set(edge_types) if edge_types is not None else None
        collections: list[str] = []
        # A walked edge is matched back to its declared edge by its collection
        # and the collections of its two ends.
        by_collection_ends: dict[tuple[str, str, str], Edge] = {}
        for edge in schema.core_schema.edge_config.edges:
            if allowed is not None and edge.relation not in allowed:
                continue
            name = edge_query_name(db_aware, edge, DBType.ARANGO)
            if name is None:
                continue
            safe = _arango_safe_collection_name(name)
            if safe not in collections:
                collections.append(safe)
            by_collection_ends[(safe, edge.source, edge.target)] = edge
        if not collections:
            return GraphContainer()

        aql_direction = {
            EdgeDirection.OUT: "OUTBOUND",
            EdgeDirection.IN: "INBOUND",
            EdgeDirection.ANY: "ANY",
        }[direction]
        limit_clause = f"LIMIT {int(limit)}" if limit is not None else ""
        # One row per reached vertex, so a cycle that reaches it by several paths
        # counts once against LIMIT; its edges are those of every such path.
        query = f"""
            FOR doc IN {anchor_collection}
                FILTER {anchor_filter}
                LIMIT 1
                FOR v, e, p IN 1..{int(hops)} {aql_direction} doc {", ".join(collections)}
                    FILTER v._id != doc._id
                    COLLECT far_id = v._id INTO walked = {{v: v, edges: p.edges}}
                    {limit_clause}
                    RETURN {{
                        far: FIRST(walked).v,
                        edges: (
                            FOR x IN UNIQUE(FLATTEN(walked[*].edges))
                                RETURN {{
                                    edge: x,
                                    start: DOCUMENT(x._from),
                                    end: DOCUMENT(x._to)
                                }}
                        )
                    }}
        """
        rows = get_data_from_cursor(self.conn.aql.execute(query, bind_vars=bind_vars))

        # A reached document's type is the collection it lives in, read from its
        # ``_id`` (``collection/key``) before internal keys are stripped.
        by_collection = {
            db_aware.vertex_config.vertex_dbname(name): name
            for name in schema.core_schema.vertex_config.vertex_set
        }

        def type_of(doc: Any) -> str | None:
            if not isinstance(doc, dict):
                return None
            return by_collection.get(str(doc.get("_id", "")).split("/", 1)[0])

        container = GraphContainer()
        seen: set[tuple[str, str]] = set()
        seen_edges: set[tuple[EdgeId, str]] = set()
        for row in rows:
            if not isinstance(row, dict):
                continue
            for walked in row.get("edges") or []:
                if limit is not None and len(seen_edges) >= limit:
                    break
                found = walked.get("edge") or {}
                start_type = type_of(walked.get("start"))
                end_type = type_of(walked.get("end"))
                if start_type is None or end_type is None:
                    continue
                edge = by_collection_ends.get(
                    (str(found.get("_id", "")).split("/", 1)[0], start_type, end_type)
                )
                if edge is None:
                    continue
                marker = (edge.edge_id, _row_marker(found, None, None))
                if marker in seen_edges:
                    continue
                seen_edges.add(marker)
                source = _vertex_identity_value(
                    self, schema, start_type, strip_internal_properties(walked["start"])
                )
                target = _vertex_identity_value(
                    self, schema, end_type, strip_internal_properties(walked["end"])
                )
                container.edges.setdefault(edge.edge_id, []).append(
                    {
                        **strip_internal_properties(found),
                        "source": source,
                        "target": target,
                    }
                )
            doc = row.get("far")
            far_type = type_of(doc)
            if far_type is None:
                continue
            cleaned = strip_internal_properties(doc)
            identity = _vertex_identity_value(self, schema, far_type, cleaned)
            if identity is None or (far_type, identity) in seen:
                continue
            seen.add((far_type, identity))
            container.vertices.setdefault(far_type, []).append(cleaned)
        container.pick_unique()
        return container

    def fetch_all_docs(
        self,
        class_name: str,
        *,
        limit: int | None = None,
    ) -> list[dict[str, Any]]:
        """Fetch all documents from a vertex collection."""
        safe = _arango_safe_collection_name(class_name)
        limit_clause = f"LIMIT {limit}" if limit is not None else ""
        query = f"""
            FOR doc IN {safe}
                {limit_clause}
                RETURN UNSET(doc, '_id', '_rev')
        """
        rows = get_data_from_cursor(self.execute(query))
        return [strip_internal_properties(row) for row in rows if isinstance(row, dict)]

    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 collections."""
        coll = collection_name or f"{source_class}_{target_class}_edges"
        safe = _arango_safe_collection_name(coll)
        limit_clause = f"LIMIT {limit}" if limit is not None else ""
        query = f"""
            FOR edge IN {safe}
                FILTER STARTS_WITH(edge._from, CONCAT('{source_class}/'))
                FILTER STARTS_WITH(edge._to, CONCAT('{target_class}/'))
                {limit_clause}
                LET source = DOCUMENT(edge._from)
                LET target = DOCUMENT(edge._to)
                RETURN {{
                    source: UNSET(source, '_id', '_rev'),
                    target: UNSET(target, '_id', '_rev'),
                    weight: UNSET(edge, '_id', '_key', '_rev', '_from', '_to')
                }}
        """
        rows = get_data_from_cursor(self.execute(query))
        result: list[list[dict[str, Any]]] = []
        for row in rows:
            if not isinstance(row, dict):
                continue
            source_doc = strip_internal_properties(row.get("source") or {})
            target_doc = strip_internal_properties(row.get("target") or {})
            edge_props = strip_internal_properties(row.get("weight") or {})
            if relation_name and "relation" not in edge_props:
                edge_props["relation"] = relation_name
            result.append([source_doc, target_doc, edge_props])
        return result

Attributes

client = ArangoClient(hosts=config.url, request_timeout=config.request_timeout) instance-attribute

config = config instance-attribute

conn = self.client.db(config.database, username=username, password=password) instance-attribute

flavor = DBType.ARANGO class-attribute instance-attribute

supports_bulk_load = True class-attribute instance-attribute

supports_graph_export = True class-attribute instance-attribute

supports_instance_delete = True class-attribute instance-attribute

supports_schema_ddl = True class-attribute instance-attribute

supports_schema_introspection = True class-attribute instance-attribute

Methods:

__init__(config)

Initialize ArangoDB connection.

Parameters:

Name	Type	Description	Default
config	ArangoConfig	ArangoDB connection configuration containing URL, credentials, and database name	required

Source code in graflo/db/arango/conn.py

def __init__(self, config: ArangoConfig):
    """Initialize ArangoDB connection.

    Args:
        config: ArangoDB connection configuration containing URL, credentials,
            and database name
    """
    super().__init__()
    # Store config for later use
    self.config = config
    # Validate required config values
    if config.url is None:
        raise ValueError("ArangoDB connection requires a URL to be configured")
    if config.database is None:
        raise ValueError(
            "ArangoDB connection requires a database name to be configured"
        )

    # ArangoDB accepts empty string for password if None
    password = config.password if config.password is not None else ""
    # ArangoDB has default username "root" if None
    username = config.username if config.username is not None else "root"

    # Store client for system operations
    self.client = ArangoClient(
        hosts=config.url, request_timeout=config.request_timeout
    )
    # Connect to the configured database for regular operations
    self.conn = self.client.db(
        config.database,
        username=username,
        password=password,
    )
    # Store credentials for system operations
    self._username = username
    self._password = password

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

Perform aggregation on a collection.

Parameters:

Name	Type	Description	Default
class_name	str	Collection to aggregate	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	None | FilterExpression | list[Any] | dict[str, Any]	Query filters	None

Returns:

Name	Type	Description
list	list[dict[str, Any]]	Aggregation results

Source code in graflo/db/arango/conn.py

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

    Args:
        class_name: Collection to aggregate
        aggregation_function: Type of aggregation to perform
        discriminant: Field to group by
        aggregated_field: Field to aggregate
        filters: Query filters

    Returns:
        list: Aggregation results
    """
    params = BoundParams(ExpressionFlavor.AQL)
    filter_clause = render_filters(filters, doc_name="doc", params=params)

    if (
        aggregated_field is not None
        and aggregation_function != AggregationType.COUNT
    ):
        group_unit = f"g[*].doc.{aggregated_field}"
    else:
        group_unit = "g"

    if discriminant is not None:
        collect_clause = f"COLLECT value = doc['{discriminant}'] INTO g"
        return_clause = f"""{{ '{discriminant}' : value, '_value': {aggregation_function}({group_unit})}}"""
    else:
        if (
            aggregated_field is None
            and aggregation_function == AggregationType.COUNT
        ):
            collect_clause = (
                f"COLLECT AGGREGATE value =  {aggregation_function} (doc)"
            )
        else:
            collect_clause = (
                "COLLECT AGGREGATE value ="
                f" {aggregation_function}(doc['{aggregated_field}'])"
            )
        return_clause = """{ '_value' : value }"""

    q = f"""FOR doc IN {class_name}
                {filter_clause}
                {collect_clause}
                RETURN {return_clause}"""

    cursor = self.execute(q, bind_vars=params.values)
    data = get_data_from_cursor(cursor)
    return data

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

Define collections, graphs, and indexes in the connected database.

Source code in graflo/db/arango/conn.py

def apply_target_schema(
    self,
    schema: Schema,
    *,
    recreate: bool,
    create_namespace: bool = True,
) -> None:
    """Define collections, graphs, and indexes in the connected database."""
    self.report_edge_direction_support(schema)
    db_name = self._resolve_db_name(schema)
    try:
        if self._schema_has_artifacts() and not recreate:
            raise SchemaExistsError(
                f"Schema already exists in database '{db_name}'. "
                "Set recreate_schema=True to replace, or use clear_data=True "
                "before ingestion."
            )
        if recreate:
            try:
                self.delete_graph_structure((), (), delete_all=True)
                logger.debug("Cleaned database '%s' for fresh start", db_name)
            except Exception as clean_error:
                logger.warning(
                    "Error during recreate for database '%s': %s",
                    db_name,
                    clean_error,
                    exc_info=True,
                )
        self.define_schema(schema)
        self.define_indexes(schema)
    except SchemaExistsError:
        raise
    except Exception as e:
        logger.error(
            "Error applying schema to database '%s': %s",
            db_name,
            e,
            exc_info=True,
        )
        raise

bulk_begin(schema, cfg, *, recreate=False, resume=False)

Guard the first write and create the collections, without their indexes.

Identity and secondary indexes are built by :meth:bulk_finish; the primary index on _key is all an import needs.

Raises:

Type	Description
SchemaExistsError	The database holds collections or graphs and neither recreate nor resume is set.

Source code in graflo/db/arango/conn.py

def bulk_begin(
    self,
    schema: Schema,
    cfg: BulkConfig,
    *,
    recreate: bool = False,
    resume: bool = False,
) -> BulkPlan:
    """Guard the first write and create the collections, without their indexes.

    Identity and secondary indexes are built by :meth:`bulk_finish`; the
    primary index on ``_key`` is all an import needs.

    Raises:
        SchemaExistsError: The database holds collections or graphs and
            neither *recreate* nor *resume* is set.
    """
    assert_schema_supported(self.flavor, schema)
    self.report_edge_direction_support(schema)
    self.ensure_target_namespace(schema, create=True)
    if not resume:
        if recreate:
            self.delete_graph_structure((), (), delete_all=True)
        elif self._schema_has_artifacts():
            raise SchemaExistsError(
                f"Schema already exists in database "
                f"'{self._resolve_db_name(schema)}'. Set recreate_schema=True "
                "to replace it."
            )
    self.define_schema(schema, identity_indexes=False)
    return BulkPlan()

bulk_finish(schema)

Build the identity and secondary indexes; count what the database holds.

Source code in graflo/db/arango/conn.py

def bulk_finish(self, schema: Schema) -> BulkCounts:
    """Build the identity and secondary indexes; count what the database holds."""
    vc = schema.resolve_db_aware(DBType.ARANGO).vertex_config
    for vertex in vc.vertex_set:
        index = vc.index(vertex)
        if index.fields and index.fields != ["_key"]:
            # Idempotent: ArangoDB returns an index that already exists.
            self._add_index(self.conn.collection(vc.vertex_dbname(vertex)), index)
    self.define_indexes(schema)
    vertices = {
        vc.vertex_dbname(vertex): int(
            cast(int, self.conn.collection(vc.vertex_dbname(vertex)).count())
        )
        for vertex in vc.vertex_set
    }
    edges: dict[StoredEdgeId, int] = {}
    for collection, stored in edge_collections(schema).items():
        if len(stored) == 1 and stored[0][1] is not None:
            count = cast(int, self.conn.collection(collection).count())
            edges[stored[0]] = edges.get(stored[0], 0) + int(count)
            continue
        # Several edges, or a relation set per row, share the collection.
        query = edge_count_query(_arango_safe_collection_name(collection))
        for source, relation, target, n in get_data_from_cursor(
            self.execute(query)
        ):
            key = (source, relation, target)
            edges[key] = edges.get(key, 0) + int(n)
    return BulkCounts(vertices=vertices, edges=edges)

bulk_prepare_edges(schema)

Source code in graflo/db/arango/conn.py

def bulk_prepare_edges(self, schema: Schema) -> None:
    """Nothing: edges carry ``_from`` and ``_to`` built from endpoint keys."""

bulk_write_edges(schema, chunk)

Import the chunk's edges, each under the _key its identity gives.

Source code in graflo/db/arango/conn.py

def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
    """Import the chunk's edges, each under the ``_key`` its identity gives."""
    docs = edge_documents(
        chunk.rows,
        chunk.source_class,
        chunk.target_class,
        chunk.match_keys_source,
        chunk.match_keys_target,
        chunk.relation_name,
        chunk.merge_properties,
        strict=True,
        fuse=False,
    )
    collection = (
        chunk.collection_name or f"{chunk.source_class}_{chunk.target_class}_edges"
    )
    self._bulk_import(collection, docs, replay=chunk.replay)

bulk_write_vertices(schema, chunk)

Import the chunk's documents, each under the _key its identity gives.

Source code in graflo/db/arango/conn.py

def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
    """Import the chunk's documents, each under the ``_key`` its identity gives."""
    docs = vertex_documents(chunk.rows, chunk.match_keys, strict=True)
    self._bulk_import(chunk.class_name, docs, replay=chunk.replay)

clear_data(schema)

Remove all data from collections without dropping the schema.

Truncates vertex and edge collections that belong to the schema.

Source code in graflo/db/arango/conn.py

def clear_data(self, schema: Schema) -> None:
    """Remove all data from collections without dropping the schema.

    Truncates vertex and edge collections that belong to the schema.
    """
    db_schema = schema.resolve_db_aware(DBType.ARANGO)
    vc = db_schema.vertex_config
    for v in vc.vertex_set:
        cname = vc.vertex_dbname(v)
        if self.conn.has_collection(cname):
            self.conn.collection(cname).truncate()
            logger.debug(f"Truncated vertex collection '{cname}'")
    for edge in db_schema.edge_config.values():
        variants = db_schema.edge_config.runtime(edge).physical_variants()
        for variant in variants:
            cname = cast(str | None, variant.get("storage_name"))
            if cname and self.conn.has_collection(cname):
                self.conn.collection(cname).truncate()
                logger.debug(f"Truncated edge collection '{cname}'")

close()

Close the ArangoDB connection.

Source code in graflo/db/arango/conn.py

def close(self) -> None:
    """Close the ArangoDB connection."""

create_collection(db_class_name, index=None, g=None)

Create a new vertex or edge class (ArangoDB uses collections internally).

Parameters:

Name	Type	Description	Default
db_class_name	str	Name of the vertex/edge class to create (ArangoDB collection name)	required
index	None | Index	Optional index to create on the class	None
g	Graph | None	Optional graph to create the class in	None

Returns:

Name	Type	Description
IndexHandle	Any | None	Handle to the created index if one was created, None otherwise

Source code in graflo/db/arango/conn.py

def create_collection(
    self,
    db_class_name: str,
    index: None | Index = None,
    g: Graph | None = None,
) -> Any | None:
    """Create a new vertex or edge class (ArangoDB uses collections internally).

    Args:
        db_class_name: Name of the vertex/edge class to create (ArangoDB collection name)
        index: Optional index to create on the class
        g: Optional graph to create the class in

    Returns:
        IndexHandle: Handle to the created index if one was created, None otherwise
    """
    if not self.conn.has_collection(db_class_name):
        if g is not None:
            _ = g.create_vertex_collection(db_class_name)
        else:
            self.conn.create_collection(db_class_name)
        general_collection = self.conn.collection(db_class_name)
        if index is not None and index.fields != ["_key"]:
            ih = self._add_index(general_collection, index)
            return ih
        else:
            return None

create_database(name)

Create a new ArangoDB database.

Database creation/deletion operations must be performed from the _system database.

Parameters:

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

Source code in graflo/db/arango/conn.py

def create_database(self, name: str) -> None:
    """Create a new ArangoDB database.

    Database creation/deletion operations must be performed from the _system database.

    Args:
        name: Name of the database to create
    """
    try:
        # Connect to _system database for system operations
        system_db = self.client.db(
            "_system", username=self._username, password=self._password
        )
        if not system_db.has_database(name):
            try:
                system_db.create_database(name)
                logger.info(f"Successfully created ArangoDB database '{name}'")
            except Exception as create_error:
                logger.error(
                    f"Failed to create ArangoDB database '{name}': {create_error}",
                    exc_info=True,
                )
                raise
        else:
            logger.debug(f"ArangoDB database '{name}' already exists")
    except Exception as e:
        logger.error(
            f"Error creating ArangoDB database '{name}': {e}",
            exc_info=True,
        )
        raise

define_edge_classes(edges, schema=None)

Define edge classes in ArangoDB.

Creates edge collections and their definitions in the appropriate graphs.

Parameters:

Name	Type	Description	Default
edges	list[Edge]	List of edge configurations to create	required
schema	Schema | None	Optional schema used to expand purpose-based physical variants	None

Source code in graflo/db/arango/conn.py

def define_edge_classes(
    self, edges: list[Edge], schema: Schema | None = None
) -> None:
    """Define edge classes in ArangoDB.

    Creates edge collections and their definitions in the appropriate graphs.

    Args:
        edges: List of edge configurations to create
        schema: Optional schema used to expand purpose-based physical variants
    """
    db_schema = (
        schema.resolve_db_aware(DBType.ARANGO) if schema is not None else None
    )
    for item in edges:
        variants = (
            db_schema.edge_config.runtime(item).physical_variants()
            if db_schema is not None
            else [
                {
                    "purpose": None,
                    "storage_name": f"{item.source}_{item.target}_edges",
                    "graph_name": f"{item.source}_{item.target}_edges",
                    "indexes": [],
                }
            ]
        )
        for variant in variants:
            gname = cast(str | None, variant.get("graph_name"))
            if not gname:
                logger.warning("Edge has no graph_name, skipping")
                continue
            if self.conn.has_graph(gname):
                g_result = self.conn.graph(gname)
            else:
                g_result = self.conn.create_graph(gname)
            # Type guard: ensure g is a Graph instance
            if not isinstance(g_result, Graph):
                logger.warning(f"Graph {gname} is not a Graph instance, skipping")
                continue
            g = g_result
            collection_name = cast(str | None, variant.get("storage_name"))
            if not collection_name:
                logger.warning("Edge has no database_name, skipping")
                continue
            if not g.has_edge_definition(collection_name):
                if db_schema is None:
                    logger.warning(
                        "DB-aware schema is required, skipping edge definition"
                    )
                    continue
                runtime = db_schema.edge_config.runtime(item)
                _ = g.create_edge_definition(
                    edge_collection=collection_name,
                    from_vertex_collections=[runtime.source_storage],
                    to_vertex_collections=[runtime.target_storage],
                )

define_edge_indexes(edges, schema=None)

Define indexes for edge collections.

Creates indexes for each edge collection based on the configuration.

Parameters:

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

Source code in graflo/db/arango/conn.py

def define_edge_indexes(
    self, edges: list[Edge], schema: Schema | None = None
) -> None:
    """Define indexes for edge collections.

    Creates indexes for each edge collection based on the configuration.

    Args:
        edges: List of edge configurations containing index definitions
    """
    if schema is None:
        logger.warning("Schema required for edge index naming, skipping")
        return
    db_schema = schema.resolve_db_aware(DBType.ARANGO)
    for edge in edges:
        variants = db_schema.edge_config.runtime(edge).physical_variants()
        for variant in variants:
            collection_name = cast(str | None, variant.get("storage_name"))
            if not collection_name:
                logger.warning("Edge has no database_name, skipping index creation")
                continue
            general_collection = self.conn.collection(collection_name)
            index_list = cast(list[Index], variant.get("indexes", []))
            for index_obj in index_list:
                self._add_index(general_collection, index_obj)

define_schema(schema, *, identity_indexes=True)

Define ArangoDB collections based on schema.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema containing collection definitions	required
identity_indexes	bool	Index each new vertex collection on its identity.	True

Source code in graflo/db/arango/conn.py

def define_schema(self, schema: Schema, *, identity_indexes: bool = True) -> None:
    """Define ArangoDB collections based on schema.

    Args:
        schema: Schema containing collection definitions
        identity_indexes: Index each new vertex collection on its identity.
    """
    assert_schema_supported(self.flavor, schema)
    self.define_vertex_classes(schema, identity_indexes=identity_indexes)
    self.define_edge_classes(
        list(schema.core_schema.edge_config.values()), schema=schema
    )

define_vertex_classes(schema, *, identity_indexes=True)

Define vertex collections in ArangoDB.

Creates vertex collections for both connected and disconnected vertices, organizing them into appropriate graphs.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema containing vertex definitions	required
identity_indexes	bool	Index each new collection on its vertex identity.	True

Source code in graflo/db/arango/conn.py

def define_vertex_classes(
    self, schema: Schema, *, identity_indexes: bool = True
) -> None:
    """Define vertex collections in ArangoDB.

    Creates vertex collections for both connected and disconnected vertices,
    organizing them into appropriate graphs.

    Args:
        schema: Schema containing vertex definitions
        identity_indexes: Index each new collection on its vertex identity.
    """
    db_schema = schema.resolve_db_aware(DBType.ARANGO)
    vertex_config = db_schema.vertex_config

    def index(vertex: str) -> Index | None:
        return vertex_config.index(vertex) if identity_indexes else None

    disconnected_vertex_collections = (
        set(vertex_config.vertex_set) - db_schema.edge_config.vertices
    )
    for item in db_schema.edge_config.values():
        u, v = item.source, item.target
        gname = db_schema.edge_config.runtime(item).graph_name()
        if not gname:
            logger.warning(
                f"Edge {item.source} -> {item.target} has no graph_name, skipping"
            )
            continue
        logger.info(f"{item.source}, {item.target}, {gname}")
        if self.conn.has_graph(gname):
            g_result = self.conn.graph(gname)
        else:
            g_result = self.conn.create_graph(gname)

        # Type narrowing: ensure g is a Graph instance
        g: Graph | None = None
        if isinstance(g_result, Graph):
            g = g_result
        elif g_result is not None:
            # If it's not a Graph, log warning and skip
            logger.warning(f"Graph {gname} is not a Graph instance, skipping")
            continue

        _ = self.create_collection(vertex_config.vertex_dbname(u), index(u), g)
        _ = self.create_collection(vertex_config.vertex_dbname(v), index(v), g)
    for v in disconnected_vertex_collections:
        _ = self.create_collection(vertex_config.vertex_dbname(v), index(v), None)

define_vertex_indexes(vertex_config, schema=None)

Define indexes for vertex collections.

Creates indexes for each vertex collection based on the configuration. Identity index is created at collection creation; this adds secondary indexes only.

Parameters:

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

Source code in graflo/db/arango/conn.py

def define_vertex_indexes(
    self, vertex_config: VertexConfig, schema: Schema | None = None
) -> None:
    """Define indexes for vertex collections.

    Creates indexes for each vertex collection based on the configuration.
    Identity index is created at collection creation; this adds secondary indexes only.

    Args:
        vertex_config: Vertex configuration containing index definitions
    """
    db_vertex = (
        schema.resolve_db_aware(DBType.ARANGO).vertex_config if schema else None
    )
    for c in vertex_config.vertex_set:
        collection_name = db_vertex.vertex_dbname(c) if db_vertex else c
        general_collection = self.conn.collection(collection_name)
        ixs = general_collection.indexes()
        field_combinations: list[tuple[Any, ...]] = []
        if isinstance(ixs, list):
            for ix in ixs:
                if isinstance(ix, dict):
                    fields_value = ix.get("fields")
                    if isinstance(fields_value, (list, tuple)):
                        field_combinations.append(tuple(fields_value))
        index_list = (
            schema.db_profile.vertex_secondary_indexes(c)
            if schema is not None
            else []
        )
        for index_obj in index_list:
            if tuple(index_obj.fields) not in field_combinations:
                self._add_index(general_collection, index_obj)

delete_database(name)

Delete an ArangoDB database.

Database creation/deletion operations must be performed from the _system database.

Parameters:

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

Source code in graflo/db/arango/conn.py

def delete_database(self, name: str) -> None:
    """Delete an ArangoDB database.

    Database creation/deletion operations must be performed from the _system database.

    Args:
        name: Name of the database to delete
    """
    try:
        # Connect to _system database for system operations
        system_db = self.client.db(
            "_system", username=self._username, password=self._password
        )
        if system_db.has_database(name):
            try:
                system_db.delete_database(name)
                logger.info(f"Successfully deleted ArangoDB database '{name}'")
            except Exception as delete_error:
                logger.error(
                    f"Failed to delete ArangoDB database '{name}': {delete_error}",
                    exc_info=True,
                )
                raise
        else:
            logger.debug(
                f"ArangoDB database '{name}' does not exist, skipping deletion"
            )
    except Exception as e:
        logger.error(
            f"Error deleting ArangoDB database '{name}': {e}",
            exc_info=True,
        )
        raise

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 edges between endpoint pairs, located by their keys.

Source code in graflo/db/arango/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 edges between endpoint pairs, located by their keys."""
    pairs = deletable_endpoints(endpoints, match_keys_source, match_keys_target)
    if not pairs:
        return
    query = arango_remove_edges_query(
        _arango_safe_collection_name(
            collection_name or f"{source_class}_{target_class}_edges"
        ),
        _arango_safe_collection_name(source_class),
        _arango_safe_collection_name(target_class),
        match_keys_source,
        match_keys_target,
        with_relation=relation_name is not None,
    )
    rows = [{"s": source, "t": target} for source, target in pairs]
    for start in range(0, len(rows), chunk_size):
        bind_vars: dict[str, Any] = {"rows": rows[start : start + chunk_size]}
        if relation_name is not None:
            bind_vars["relation"] = relation_name
        self.conn.aql.execute(query, bind_vars=bind_vars)

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

Delete graph structure (vertex/edge classes and graphs) from ArangoDB.

In ArangoDB: - Collections (internal): Container for vertices (vertex collections) and edges (edge collections) - Graphs: Named graphs that connect vertex and edge collections

Parameters:

Name	Type	Description	Default
vertex_types	tuple[str, ...] | list[str]	Vertex/edge class names to delete (ArangoDB collection names)	()
graph_names	tuple[str, ...] | list[str]	Graph names to delete	()
delete_all	bool	If True, delete all non-system vertex/edge classes and graphs	False

Source code in graflo/db/arango/conn.py

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 (vertex/edge classes and graphs) from ArangoDB.

    In ArangoDB:
    - Collections (internal): Container for vertices (vertex collections) and edges (edge collections)
    - Graphs: Named graphs that connect vertex and edge collections

    Args:
        vertex_types: Vertex/edge class names to delete (ArangoDB collection names)
        graph_names: Graph names to delete
        delete_all: If True, delete all non-system vertex/edge classes and graphs
    """

    def _graph_name(graph_item: Any) -> str | None:
        if not isinstance(graph_item, dict):
            return None
        graph_dict = cast(dict[str, Any], graph_item)
        for key in ("name", "_key"):
            value = graph_dict.get(key)
            if isinstance(value, str) and value:
                return value
        return None

    cnames: list[str] = list(vertex_types)
    gnames: list[str] = list(graph_names)
    logger.info("vertex/edge classes (non system, ArangoDB collections):")
    collections_result = self.conn.collections()
    if isinstance(collections_result, list):
        filtered_collections: list[dict[str, Any]] = []
        for c in collections_result:
            if isinstance(c, dict):
                name_value = c.get("name")
                if isinstance(name_value, str) and name_value[0] != "_":
                    filtered_collections.append(c)
        logger.info(filtered_collections)
    else:
        logger.info([])

    if delete_all:
        collections_result = self.conn.collections()
        graphs_result = self.conn.graphs()
        cnames = []
        if isinstance(collections_result, list):
            for c in collections_result:
                if isinstance(c, dict):
                    name_value = c.get("name")
                    if isinstance(name_value, str) and name_value[0] != "_":
                        cnames.append(name_value)
        gnames = []
        if isinstance(graphs_result, list):
            for g in graphs_result:
                name_value = _graph_name(g)
                if name_value is not None:
                    gnames.append(name_value)

    # Delete graphs first. Retry because graph metadata can be stale after deletes.
    for _ in range(3):
        deleted_any = False
        for gn in gnames:
            if self.conn.has_graph(gn):
                self.conn.delete_graph(gn)
                deleted_any = True
        if not deleted_any:
            break
        # Refresh remaining graph names before next pass.
        refreshed_graphs = self.conn.graphs()
        gnames = []
        if isinstance(refreshed_graphs, list):
            for g in refreshed_graphs:
                name_value = _graph_name(g)
                if name_value is not None:
                    gnames.append(name_value)

    logger.info("graphs (after delete operation):")
    logger.info(self.conn.graphs())

    # Delete collections. If any collection is still attached to a graph,
    # remove graphs again and retry once.
    for attempt in range(2):
        blocked = False
        for cn in cnames:
            if not self.conn.has_collection(cn):
                continue
            try:
                self.conn.delete_collection(cn)
            except Exception as e:
                if "ERR 1942" in str(e) and attempt == 0:
                    blocked = True
                    continue
                raise
        if not blocked:
            break
        refreshed_graphs = self.conn.graphs()
        if isinstance(refreshed_graphs, list):
            for g in refreshed_graphs:
                gn = _graph_name(g)
                if gn is not None and self.conn.has_graph(gn):
                    self.conn.delete_graph(gn)

    logger.info(
        "vertex/edge classes (after delete operation, ArangoDB collections):"
    )
    collections_result = self.conn.collections()
    if isinstance(collections_result, list):
        collection_names: list[str] = []
        for c in collections_result:
            if isinstance(c, dict):
                name_value = c.get("name")
                if isinstance(name_value, str) and name_value[0] != "_":
                    collection_names.append(name_value)
        logger.info(collection_names)
    else:
        logger.info([])

    logger.info("graphs:")
    logger.info(self.conn.graphs())

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

Remove documents and every edge, in any edge collection, that touches them.

Source code in graflo/db/arango/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 documents and every edge, in any edge collection, that touches them."""
    rows = deletable_docs(key_docs, match_keys)
    if not rows:
        return
    edge_collections = self._edge_collection_names()
    collection = _arango_safe_collection_name(class_name)
    for start in range(0, len(rows), chunk_size):
        removed = list(
            get_data_from_cursor(
                self.conn.aql.execute(
                    arango_remove_documents_query(collection, match_keys),
                    bind_vars={"rows": rows[start : start + chunk_size]},
                )
            )
        )
        if not removed:
            continue
        for edges in edge_collections:
            self.conn.aql.execute(
                arango_remove_incident_edges_query(
                    _arango_safe_collection_name(edges)
                ),
                bind_vars={"ids": removed},
            )

ensure_target_namespace(schema, *, create)

Ensure the ArangoDB database exists and connect to it.

Source code in graflo/db/arango/conn.py

def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
    """Ensure the ArangoDB database exists and connect to it."""
    db_name = self._resolve_db_name(schema)
    try:
        if not self._database_exists(db_name):
            if not create:
                raise NamespaceNotFoundError(
                    f"ArangoDB database '{db_name}' does not exist. "
                    "Create it manually or call with create_namespace=True."
                )
            logger.info("Database '%s' does not exist, creating it...", db_name)
            system_db = self.client.db(
                "_system", username=self._username, password=self._password
            )
            system_db.create_database(db_name)
            logger.info("Successfully created database '%s'", db_name)
        self._connect_to_database(db_name)
    except NamespaceNotFoundError:
        raise
    except Exception as e:
        logger.error("Error ensuring database '%s': %s", db_name, e, exc_info=True)
        raise

execute(query, bind_vars=None, **kwargs)

Execute an AQL query.

Parameters:

Name	Type	Description	Default
query	str	AQL query string to execute	required
bind_vars	dict[str, Any] | None	Values the query names as @name	None
**kwargs	Any	Additional query parameters	{}

Returns:

Name	Type	Description
Cursor	Any	ArangoDB cursor for the query results

Source code in graflo/db/arango/conn.py

def execute(
    self, query: str, bind_vars: dict[str, Any] | None = None, **kwargs: Any
) -> Any:
    """Execute an AQL query.

    Args:
        query: AQL query string to execute
        bind_vars: Values the query names as ``@name``
        **kwargs: Additional query parameters

    Returns:
        Cursor: ArangoDB cursor for the query results
    """
    cursor = self.conn.aql.execute(query, bind_vars=bind_vars or None)
    return cursor

fetch_all_docs(class_name, *, limit=None)

Fetch all documents from a vertex collection.

Source code in graflo/db/arango/conn.py

def fetch_all_docs(
    self,
    class_name: str,
    *,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """Fetch all documents from a vertex collection."""
    safe = _arango_safe_collection_name(class_name)
    limit_clause = f"LIMIT {limit}" if limit is not None else ""
    query = f"""
        FOR doc IN {safe}
            {limit_clause}
            RETURN UNSET(doc, '_id', '_rev')
    """
    rows = get_data_from_cursor(self.execute(query))
    return [strip_internal_properties(row) for row in rows if isinstance(row, dict)]

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

Source code in graflo/db/arango/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 collections."""
    coll = collection_name or f"{source_class}_{target_class}_edges"
    safe = _arango_safe_collection_name(coll)
    limit_clause = f"LIMIT {limit}" if limit is not None else ""
    query = f"""
        FOR edge IN {safe}
            FILTER STARTS_WITH(edge._from, CONCAT('{source_class}/'))
            FILTER STARTS_WITH(edge._to, CONCAT('{target_class}/'))
            {limit_clause}
            LET source = DOCUMENT(edge._from)
            LET target = DOCUMENT(edge._to)
            RETURN {{
                source: UNSET(source, '_id', '_rev'),
                target: UNSET(target, '_id', '_rev'),
                weight: UNSET(edge, '_id', '_key', '_rev', '_from', '_to')
            }}
    """
    rows = get_data_from_cursor(self.execute(query))
    result: list[list[dict[str, Any]]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        source_doc = strip_internal_properties(row.get("source") or {})
        target_doc = strip_internal_properties(row.get("target") or {})
        edge_props = strip_internal_properties(row.get("weight") or {})
        if relation_name and "relation" not in edge_props:
            edge_props["relation"] = relation_name
        result.append([source_doc, target_doc, edge_props])
    return result

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

Fetch documents from a collection.

Parameters:

Name	Type	Description	Default
class_name	str	Collection to fetch from	required
filters	None | FilterExpression | list[Any] | dict[str, Any]	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

Returns:

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

Source code in graflo/db/arango/conn.py

def fetch_docs(
    self,
    class_name: str,
    filters: None | FilterExpression | list[Any] | dict[str, Any] = 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 collection.

    Args:
        class_name: Collection to fetch from
        filters: Query filters
        limit: Maximum number of documents to return
        return_keys: Keys to return
        unset_keys: Keys to unset

    Returns:
        list: Fetched documents
    """
    params = BoundParams(ExpressionFlavor.AQL)
    filter_clause = render_filters(filters, doc_name="d", params=params)

    if return_keys is None:
        if unset_keys is None:
            return_clause = "d"
        else:
            tmp_clause = ", ".join([f'"{item}"' for item in unset_keys])
            return_clause = f"UNSET(d, {tmp_clause})"
    else:
        if unset_keys is None:
            tmp_clause = ", ".join([f'"{item}"' for item in return_keys])
            return_clause = f"KEEP(d, {tmp_clause})"
        else:
            raise ValueError("both return_keys and unset_keys are set")

    if limit is not None and isinstance(limit, int):
        limit_clause = f"LIMIT {limit}"
    else:
        limit_clause = ""

    q = (
        f"FOR d in {class_name}"
        f"  {filter_clause}"
        f"  {limit_clause}"
        f"  RETURN {return_clause}"
    )
    cursor = self.execute(q, bind_vars=params.values)
    return get_data_from_cursor(cursor)

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)

Fetch edges from ArangoDB using AQL.

Reverse lookup is free here: an edge collection's edge index covers _from and _to, so IN and ANY cost the same as OUT.

Parameters:

Name	Type	Description	Default
from_type	str	Anchor vertex collection name	required
from_id	str	Anchor vertex ID (can be _key or _id)	required
edge_type	str | None	Optional edge collection name to filter by	None
to_type	str | None	Optional collection of the other endpoint	None
to_id	str | None	Optional ID of the other endpoint	None
filters	list[Any] | dict[str, Any] | FilterExpression | 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	OUT
**kwargs	Any	Additional parameters	{}

Returns:

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

Source code in graflo/db/arango/conn.py

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] | FilterExpression | 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 from ArangoDB using AQL.

    Reverse lookup is free here: an edge collection's edge index covers
    ``_from`` *and* ``_to``, so ``IN`` and ``ANY`` cost the same as ``OUT``.

    Args:
        from_type: Anchor vertex collection name
        from_id: Anchor vertex ID (can be _key or _id)
        edge_type: Optional edge collection name to filter by
        to_type: Optional collection of the other endpoint
        to_id: Optional 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
        **kwargs: Additional parameters

    Returns:
        list: List of fetched edges
    """
    # Convert from_id to _id format if needed
    if not from_id.startswith(from_type):
        # Assume it's a _key, convert to _id
        from_vertex_id = f"{from_type}/{from_id}"
    else:
        from_vertex_id = from_id

    # Build AQL query to fetch edges
    # Start with basic edge traversal
    if edge_type:
        edge_collection = edge_type
    else:
        # If no edge_type specified, we need to search all edge collections
        # This is a simplified version - in practice you might want to list all edge collections
        raise ValueError("edge_type is required for ArangoDB edge fetching")

    params = BoundParams(ExpressionFlavor.AQL)
    # Already a `FILTER ...` clause, or empty.
    filter_clause = render_filters(filters, doc_name="e", params=params)

    to_vertex_id: str | None = None
    if to_id and to_type:
        to_vertex_id = to_id if to_id.startswith(to_type) else f"{to_type}/{to_id}"

    bind_vars: dict[str, Any] = {}
    anchor_clause = _arango_edge_anchor_clause(
        from_vertex_id, direction, to_type, to_vertex_id, bind_vars
    )
    bind_vars.update(params.values)

    query = f"""
        FOR e IN {edge_collection}
            FILTER {anchor_clause}
            {filter_clause}
            {f"LIMIT {limit}" if limit else ""}
            RETURN e
    """

    cursor = self.execute(query, bind_vars=bind_vars)
    result = list(get_data_from_cursor(cursor))

    # Apply projection
    if return_keys is not None:
        result = [
            {k: doc.get(k) for k in return_keys if k in doc} for doc in result
        ]
    elif unset_keys is not None:
        result = [
            {k: v for k, v in doc.items() if k not in unset_keys} for doc in result
        ]

    return result

fetch_indexes(db_class_name=None)

Fetch all indexes from the database.

Parameters:

Name	Type	Description	Default
db_class_name	str | None	Optional collection name to fetch indexes for	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Mapping of collection names to their indexes

Source code in graflo/db/arango/conn.py

def fetch_indexes(self, db_class_name: str | None = None) -> dict[str, Any]:
    """Fetch all indexes from the database.

    Args:
        db_class_name: Optional collection name to fetch indexes for

    Returns:
        dict: Mapping of collection names to their indexes
    """
    classes: list[Any] = []
    if db_class_name is None:
        classes_result = self.conn.collections()
        if isinstance(classes_result, list):
            classes = classes_result
    elif self.conn.has_collection(db_class_name):
        classes = [self.conn.collection(db_class_name)]

    r: dict[str, Any] = {}
    for cname in classes:
        if isinstance(cname, dict):
            cname_dict = cast(dict[str, Any], cname)
            name_value = cname_dict.get("name")
            if isinstance(name_value, str):
                c = self.conn.collection(name_value)
                r[name_value] = c.indexes()
    return r

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

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	Collection to check in	required
match_keys	list[str] | tuple[str, ...]	Keys to match documents	required
keep_keys	list[str] | tuple[str, ...] | None	Keys to keep in result	None
flatten	bool	If True, flatten the result into a list	False
filters	None | FilterExpression | list[Any] | dict[str, Any]	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, either as a flat list or a dictionary mapping batch indices to documents

Source code in graflo/db/arango/conn.py

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: None | FilterExpression | list[Any] | dict[str, Any] = 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: Collection to check in
        match_keys: Keys to match documents
        keep_keys: Keys to keep in result
        flatten: If True, flatten the result into a list
        filters: Additional query filters

    Returns:
        list | dict: Documents that exist in the database, either as a
            flat list or a dictionary mapping batch indices to documents
    """
    q0, bind_vars = fetch_fields_query(
        collection_name=class_name,
        docs=batch,
        match_keys=match_keys,
        keep_keys=keep_keys,
        filters=filters,
    )
    # {"__i": i, "_group": [doc]}
    cursor = self.execute(q0, bind_vars=bind_vars)

    if flatten:
        rdata = []
        for item in get_data_from_cursor(cursor):
            group = item.pop("_group", [])
            rdata += [sub_item for sub_item in group]
        return rdata
    else:
        rdata_dict: dict[int, list[dict[str, Any]]] = {}
        for item in get_data_from_cursor(cursor):
            __i = item.pop("__i")
            group = item.pop("_group", [])
            rdata_dict[__i] = group
        return rdata_dict

get_collections()

Get all vertex and edge classes in the database (ArangoDB collections).

Returns:

Name	Type	Description
list	list[dict[str, Any]]	List of class information dictionaries (ArangoDB collection info)

Source code in graflo/db/arango/conn.py

def get_collections(self) -> list[dict[str, Any]]:
    """Get all vertex and edge classes in the database (ArangoDB collections).

    Returns:
        list: List of class information dictionaries (ArangoDB collection info)
    """
    result = self.conn.collections()
    if isinstance(result, list):
        return [item if isinstance(item, dict) else {} for item in result]
    return []

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

Convenience wrapper: ensure database then apply schema.

Source code in graflo/db/arango/conn.py

def init_db(
    self,
    schema: Schema,
    recreate_schema: bool = False,
    *,
    create_namespace: bool = True,
) -> None:
    """Convenience wrapper: ensure database then apply schema."""
    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=('_key',), match_keys_target=('_key',), filter_uniques=True, head=None, **kwargs)

Upsert a batch of edges on their _key.

An edge is its endpoints, relation and identity fields, digested into its _key; a row with the same identity updates the edge's other properties. _from and _to are built from the endpoints' match key values, so no vertex is looked up and an absent endpoint is not checked. Rows missing an endpoint value are dropped.

Parameters:

Name	Type	Description	Default
docs_edges	list[list[dict[str, Any]]] | list[Any] | None	[source_doc, target_doc, props] rows	required
source_class	str	Source vertex collection	required
target_class	str	Target vertex collection	required
relation_name	str	Relation stored on each edge, if any	required
match_keys_source	tuple[str, ...]	The source's identity fields, in identity order	('_key',)
match_keys_target	tuple[str, ...]	The target's identity fields, in identity order	('_key',)
filter_uniques	bool	If True, filter duplicate rows	True
head	int | None	Optional limit on number of edges to insert	None
**kwargs	Any	Additional options: - dry: If True, don't execute the query - collection_name: Edge collection name (defaults to {source_class}_{target_class}_edges if not provided) - uniq_weight_fields: The edge's identity fields (parallel edges) - uniq_weight_collections: Extra identity fields (legacy; list of names) - relationship_merge_properties: Ignored (Cypher backends only)	{}

Source code in graflo/db/arango/conn.py

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, ...] = ("_key",),
    match_keys_target: tuple[str, ...] = ("_key",),
    filter_uniques: bool = True,
    head: int | None = None,
    **kwargs: Any,
) -> None:
    """Upsert a batch of edges on their ``_key``.

    An edge is its endpoints, relation and identity fields, digested into
    its ``_key``; a row with the same identity updates the edge's other
    properties. ``_from`` and ``_to`` are built from the endpoints' match
    key values, so no vertex is looked up and an absent endpoint is not
    checked. Rows missing an endpoint value are dropped.

    Args:
        docs_edges: ``[source_doc, target_doc, props]`` rows
        source_class: Source vertex collection
        target_class: Target vertex collection
        relation_name: Relation stored on each edge, if any
        match_keys_source: The source's identity fields, in identity order
        match_keys_target: The target's identity fields, in identity order
        filter_uniques: If True, filter duplicate rows
        head: Optional limit on number of edges to insert
        **kwargs: Additional options:
            - dry: If True, don't execute the query
            - collection_name: Edge collection name (defaults to {source_class}_{target_class}_edges if not provided)
            - uniq_weight_fields: The edge's identity fields (parallel edges)
            - uniq_weight_collections: Extra identity fields (legacy; list of names)
            - relationship_merge_properties: Ignored (Cypher backends only)
    """
    opts = consume_insert_edges_kwargs(kwargs)
    collection_name = opts.collection_name or f"{source_class}_{target_class}_edges"

    if not isinstance(docs_edges, list):
        return
    if head is not None:
        docs_edges = docs_edges[:head]
    if filter_uniques:
        docs_edges = pick_unique_dict(docs_edges)
    docs = edge_documents(
        docs_edges,
        source_class,
        target_class,
        match_keys_source,
        match_keys_target,
        relation_name,
        _arango_edge_uniq_weight_keys(
            opts.uniq_weight_fields, opts.uniq_weight_collections
        ),
        strict=False,
        fuse=True,
    )
    if not docs or opts.dry:
        return
    self.execute(
        arango_upsert_edges_query(
            collection_name, json.dumps(docs, default=json_serializer)
        )
    )

insert_return_batch(docs, class_name)

Insert documents and return the AQL query string.

Note: ArangoDB-specific behavior - returns query string instead of executing. This allows for deferred execution and is used by caster.py and tests.

Parameters:

Name	Type	Description	Default
docs	list[dict[str, Any]]	Documents to insert	required
class_name	str	Collection to insert into	required

Returns:

Name	Type	Description
str	str	AQL query string for the operation (can be executed with execute())

Source code in graflo/db/arango/conn.py

def insert_return_batch(self, docs: list[dict[str, Any]], class_name: str) -> str:
    """Insert documents and return the AQL query string.

    Note: ArangoDB-specific behavior - returns query string instead of executing.
    This allows for deferred execution and is used by caster.py and tests.

    Args:
        docs: Documents to insert
        class_name: Collection to insert into

    Returns:
        str: AQL query string for the operation (can be executed with execute())
    """
    docs_str = json.dumps(docs, default=json_serializer)
    query0 = f"""FOR doc in {docs_str}
          INSERT doc
          INTO {class_name}
          LET inserted = NEW
          RETURN {{_key: inserted._key}}
    """
    return query0

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

Infer a graflo Schema from ArangoDB via collection sampling.

Source code in graflo/db/arango/conn.py

def introspect_graph_schema(
    self,
    schema_name: str | None = None,
    *,
    sample_limit: int = 100,
) -> Schema:
    """Infer a graflo Schema from ArangoDB via collection sampling."""
    resolved_name = schema_name or self.config.database or "arango"
    collections = self.get_collections()
    vertex_collections: list[str] = []
    edge_collections: list[str] = []
    for coll in collections:
        name = coll.get("name")
        if not name or name.startswith("_"):
            continue
        kind = _arango_collection_kind(coll.get("type"))
        if kind == "edge":
            edge_collections.append(name)
        elif kind == "document":
            vertex_collections.append(name)

    vertices: list[GraphVertexIntrospection] = []
    for coll_name in vertex_collections:
        safe = _arango_safe_collection_name(coll_name)
        props_query = f"""
            FOR doc IN {safe}
                LIMIT {sample_limit}
                FOR key IN ATTRIBUTES(doc)
                    FILTER key NOT IN ['_id', '_key', '_rev']
                    RETURN DISTINCT key
        """
        prop_rows = get_data_from_cursor(self.execute(props_query))
        properties = [row for row in prop_rows if isinstance(row, str)]
        vertices.append(
            GraphVertexIntrospection(
                name=coll_name,
                properties=properties,
                identity=infer_identity_fields(properties),
            )
        )

    edges: list[GraphEdgeIntrospection] = []
    for coll_name in edge_collections:
        safe = _arango_safe_collection_name(coll_name)
        pattern_query = f"""
            FOR edge IN {safe}
                LIMIT {sample_limit}
                COLLECT source = SPLIT(edge._from, '/')[0],
                        target = SPLIT(edge._to, '/')[0]
                RETURN {{ source, target }}
        """
        pattern_rows = get_data_from_cursor(self.execute(pattern_query))
        edge_props_query = f"""
            FOR edge IN {safe}
                LIMIT {sample_limit}
                FOR key IN ATTRIBUTES(edge)
                    FILTER key NOT IN ['_id', '_key', '_rev', '_from', '_to']
                    RETURN DISTINCT key
        """
        edge_prop_rows = get_data_from_cursor(self.execute(edge_props_query))
        edge_properties = [row for row in edge_prop_rows if isinstance(row, str)]
        relation: str | None = None
        if "relation" in edge_properties:
            relation = "relation"
        seen_pairs: set[tuple[str, str]] = set()
        for row in pattern_rows:
            if not isinstance(row, dict):
                continue
            source = row.get("source")
            target = row.get("target")
            if not source or not target:
                continue
            pair = (source, target)
            if pair in seen_pairs:
                continue
            seen_pairs.add(pair)
            edges.append(
                GraphEdgeIntrospection(
                    source=source,
                    target=target,
                    relation=relation,
                    properties=edge_properties,
                    collection_name=coll_name,
                )
            )

    introspection = GraphIntrospectionResult(
        name=resolved_name,
        vertices=vertices,
        edges=edges,
    )
    return GraphSchemaInferencer(db_flavor=DBType.ARANGO).infer_schema(
        introspection, schema_name=resolved_name
    )

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

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	Collection to check in	required
match_keys	list[str] | tuple[str, ...]	Keys to match documents	required
keep_keys	list[str] | tuple[str, ...] | None	Keys to keep in result	None
filters	None | FilterExpression | list[Any] | dict[str, Any]	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/arango/conn.py

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: None | FilterExpression | list[Any] | dict[str, Any] = None,
) -> list[dict[str, Any]]:
    """Keep documents that don't exist in the database.

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

    Returns:
        list: Documents that don't exist in the database
    """
    present_docs_keys = self.fetch_present_documents(
        batch=batch,
        class_name=class_name,
        match_keys=match_keys,
        keep_keys=keep_keys,
        flatten=False,
        filters=filters,
    )

    if not isinstance(present_docs_keys, dict):
        raise TypeError(f"expected dict, got {type(present_docs_keys).__name__}")

    if any(len(v) > 1 for v in present_docs_keys.values()):
        logger.warning(
            "fetch_present_documents returned multiple docs per filtering condition"
        )

    absent_indices = sorted(set(range(len(batch))) - set(present_docs_keys.keys()))
    batch_absent = [batch[j] for j in absent_indices]
    return batch_absent

update_to_numeric(collection_name, field)

Update a field to numeric type in all documents.

Parameters:

Name	Type	Description	Default
collection_name	str	Vertex/edge class name to update (ArangoDB collection name)	required
field	str	Field to convert to numeric	required

Returns:

Name	Type	Description
str	str	AQL query string for the operation

Source code in graflo/db/arango/conn.py

def update_to_numeric(self, collection_name: str, field: str) -> str:
    """Update a field to numeric type in all documents.

    Args:
        collection_name: Vertex/edge class name to update (ArangoDB collection name)
        field: Field to convert to numeric

    Returns:
        str: AQL query string for the operation
    """
    s1 = f"FOR p IN {collection_name} FILTER p.{field} update p with {{"
    s2 = f"{field}: TO_NUMBER(p.{field}) "
    s3 = f"}} in {collection_name}"
    q0 = s1 + s2 + s3
    return q0

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

Upsert a batch of documents using AQL.

Documents are matched on match_keys; each gets the _key those values determine (:func:graflo.db.arango.keys.vertex_key), so edges can name it without a lookup.

Parameters:

Name	Type	Description	Default
docs	list[dict[str, Any]]	List of documents to upsert	required
class_name	str	Collection name to upsert into	required
match_keys	list[str] | tuple[str, ...]	Keys to match for upsert operation	()
**kwargs	Any	Additional options: - dry: If True, don't execute the query - update_keys: Keys to update on match - filter_uniques: If True, filter duplicate documents	{}

Source code in graflo/db/arango/conn.py

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 using AQL.

    Documents are matched on *match_keys*; each gets the ``_key`` those
    values determine (:func:`graflo.db.arango.keys.vertex_key`), so edges
    can name it without a lookup.

    Args:
        docs: List of documents to upsert
        class_name: Collection name to upsert into
        match_keys: Keys to match for upsert operation
        **kwargs: Additional options:
            - dry: If True, don't execute the query
            - update_keys: Keys to update on match
            - filter_uniques: If True, filter duplicate documents
    """
    dry = kwargs.pop("dry", False)
    update_keys = kwargs.pop("update_keys", None)
    filter_uniques = kwargs.pop("filter_uniques", True)

    if not docs:
        return
    if filter_uniques:
        docs = pick_unique_dict(docs)
    docs = vertex_documents(docs, match_keys, strict=False)
    docs_json = json.dumps(docs, default=json_serializer)
    if not match_keys:
        upsert_clause = ""
        update_clause = ""
    else:
        upsert_clause = ", ".join([f'"{k}": doc.{k}' for k in match_keys])
        upsert_clause = f"UPSERT {{{upsert_clause}}}"

        if isinstance(update_keys, list):
            update_clause = ", ".join([f'"{k}": doc.{k}' for k in update_keys])
            update_clause = f"{{{update_clause}}}"
        elif update_keys == "doc":
            update_clause = "doc"
        else:
            update_clause = "{}"
        update_clause = f"UPDATE {update_clause}"

    options = "OPTIONS {ignoreErrors: true}"

    q_update = f"""FOR doc in {docs_json}
                        {upsert_clause}
                        INSERT doc
                        {update_clause}
                            IN {class_name} {options}"""
    if not dry:
        self.execute(q_update)

Functions:

arango_remove_documents_query(collection, match_keys)

AQL removing the documents that match a row of @rows; returns their _id.

Source code in graflo/db/arango/conn.py

def arango_remove_documents_query(collection: str, match_keys: Sequence[str]) -> str:
    """AQL removing the documents that match a row of ``@rows``; returns their ``_id``."""
    condition = " AND ".join(
        f"doc.{_aql_attribute(key)} == row.{_aql_attribute(key)}" for key in match_keys
    )
    return (
        f"FOR row IN @rows FOR doc IN {collection} FILTER {condition} "
        f"REMOVE doc IN {collection} RETURN OLD._id"
    )

arango_remove_edges_query(edge_collection, source_collection, target_collection, source_keys, target_keys, *, with_relation)

AQL removing the edges between the endpoint pairs of @rows.

A row is {"s": source keys, "t": target keys}. With with_relation, only edges whose relation is @relation.

Source code in graflo/db/arango/conn.py

def arango_remove_edges_query(
    edge_collection: str,
    source_collection: str,
    target_collection: str,
    source_keys: Sequence[str],
    target_keys: Sequence[str],
    *,
    with_relation: bool,
) -> str:
    """AQL removing the edges between the endpoint pairs of ``@rows``.

    A row is ``{"s": source keys, "t": target keys}``. With *with_relation*,
    only edges whose ``relation`` is ``@relation``.
    """
    source = " AND ".join(
        f"s.{_aql_attribute(k)} == row.s.{_aql_attribute(k)}" for k in source_keys
    )
    target = " AND ".join(
        f"t.{_aql_attribute(k)} == row.t.{_aql_attribute(k)}" for k in target_keys
    )
    relation = " AND e.relation == @relation" if with_relation else ""
    return (
        f"FOR row IN @rows "
        f"FOR s IN {source_collection} FILTER {source} "
        f"FOR t IN {target_collection} FILTER {target} "
        f"FOR e IN {edge_collection} FILTER e._from == s._id AND e._to == t._id{relation} "
        f"REMOVE e IN {edge_collection}"
    )

arango_remove_incident_edges_query(edge_collection)

AQL removing the edges of edge_collection with an endpoint in @ids.

Source code in graflo/db/arango/conn.py

def arango_remove_incident_edges_query(edge_collection: str) -> str:
    """AQL removing the edges of *edge_collection* with an endpoint in ``@ids``."""
    return (
        f"FOR e IN {edge_collection} FILTER e._from IN @ids OR e._to IN @ids "
        f"REMOVE e IN {edge_collection}"
    )

arango_upsert_edges_query(collection, docs_json)

AQL writing edge documents that carry their _key.

A new edge is inserted; a stored one takes the document's attributes, as UPDATE would. Keyed on _key, the write is atomic per edge.

Source code in graflo/db/arango/conn.py

def arango_upsert_edges_query(collection: str, docs_json: str) -> str:
    """AQL writing edge documents that carry their ``_key``.

    A new edge is inserted; a stored one takes the document's attributes, as
    ``UPDATE`` would. Keyed on ``_key``, the write is atomic per edge.
    """
    return (
        f"FOR doc IN {docs_json} INSERT doc INTO {collection} "
        'OPTIONS {overwriteMode: "update"}'
    )