graflo.db.neo4j.conn

Neo4j connection implementation for graph database operations.

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

• Node and relationship management
• Cypher query execution
• Index creation and management
• Batch operations
• Graph traversal and pattern matching

Key Features:

- Label-based node organization
- Relationship type management
- Property indexes
- Cypher query execution
- Batch node and relationship operations

Example

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

Attributes

DEFAULT_NEO4J_USERNAME = 'neo4j' module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Neo4jConnection

Bases: Connection

Neo4j-specific implementation of the Connection interface.

This class provides Neo4j-specific implementations for all database operations, including node management, relationship operations, and Cypher query execution. It uses the Neo4j Python driver for all operations.

Attributes:

Name	Type	Description
flavor		Database flavor identifier (NEO4J)
conn		Neo4j session instance

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

class Neo4jConnection(Connection):
    """Neo4j-specific implementation of the Connection interface.

    This class provides Neo4j-specific implementations for all database
    operations, including node management, relationship operations, and
    Cypher query execution. It uses the Neo4j Python driver for all operations.

    Attributes:
        flavor: Database flavor identifier (NEO4J)
        conn: Neo4j session instance
    """

    flavor = DBType.NEO4J
    supports_instance_delete = True
    supports_scoped_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_read_guard = True
    supports_union_merge = True
    supports_membership_read = True
    supports_bulk_load = True
    # Creating a relationship locks both endpoint nodes.
    bulk_partition_edges = True
    supports_sync = True

    def __init__(self, config: Neo4jConfig):
        """Initialize Neo4j connection.

        Args:
            config: Neo4j connection configuration containing URL and credentials
        """
        super().__init__()
        # Store config for later use
        self.config = config
        # Ensure url is not None - GraphDatabase.driver requires a non-None URI
        if config.url is None:
            raise ValueError("Neo4j connection requires a URL to be configured")
        # A password is never dropped: without a username it is Neo4j's built-in
        # user's, and only a config with no password connects unauthenticated.
        auth = None
        if config.password is not None:
            auth = (config.username or DEFAULT_NEO4J_USERNAME, config.password)
        # Queries are built from the schema, so an unrecognized label, property or
        # relation is one nothing has written yet, never a typo: no notification.
        self._driver = GraphDatabase.driver(
            uri=config.url,
            auth=auth,
            notifications_disabled_categories=[
                NotificationDisabledCategory.UNRECOGNIZED
            ],
        )
        # ``None`` is the user's home database.
        self.conn = self._driver.session(database=config.database)

    def execute(self, query, **kwargs):
        """Execute a Cypher query.

        Args:
            query: Cypher query string to execute
            **kwargs: Additional query parameters

        Returns:
            Result: Neo4j query result
        """
        cursor = self.conn.run(query, **kwargs)
        return cursor

    def reset_after_use(self) -> None:
        """Start a fresh session: closing one finishes any result left pending."""
        self.conn.close()
        self.conn = self._driver.session(database=self.config.database)

    def close(self):
        """Close the Neo4j connection and session."""
        # Close session first, then the underlying driver
        try:
            self.conn.close()
        finally:
            # Ensure the driver is also closed to release resources
            self._driver.close()

    def create_database(self, name: str):
        """Create a new Neo4j database.

        Note: This operation is only supported in Neo4j Enterprise Edition.
        Community Edition only supports one database per instance.

        Args:
            name: Name of the database to create

        Raises:
            Exception: If database creation fails and it's not a Community Edition limitation
        """
        try:
            self.execute(f"CREATE DATABASE {cypher_name(name, quoted=True)}")
            logger.info(f"Successfully created Neo4j database '{name}'")
        except Exception as e:
            # Check if this is a Neo4j Community Edition limitation
            error_str = str(e).lower()
            if (
                "unsupported administration command" in error_str
                or "create database" in error_str
            ):
                # This is likely Neo4j Community Edition - don't raise, just log
                logger.info(
                    f"Neo4j Community Edition detected: Cannot create database '{name}'. "
                    f"Community Edition only supports one database per instance. "
                    f"Continuing with default database."
                )
                # Don't raise - allow operation to continue with default database
                return
            # For other errors, raise them
            raise

    def delete_database(self, name: str):
        """Delete a Neo4j database.

        Note: This operation is only supported in Neo4j Enterprise Edition.
        As a fallback, it deletes all nodes and relationships.

        Args:
            name: Name of the database to delete (unused, deletes all data)
        """
        try:
            self._drop_all_user_indexes_and_constraints()
            self.execute("MATCH (n) DETACH DELETE n")
            logger.info("Successfully cleaned Neo4j database")
        except Exception as e:
            logger.error(
                f"Failed to clean Neo4j database: {e}",
                exc_info=True,
            )
            raise

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

        Creates indexes for each vertex label based on the configuration.
        Neo4j has no implicit identity index; the identity, as a uniqueness
        constraint, is prepended when schema is provided.

        Args:
            vertex_config: Vertex configuration containing index definitions
        """
        if schema is None:
            logger.warning(
                "Schema is None: identity indexes cannot be ensured without schema"
            )
        db_vertex = (
            schema.resolve_db_aware(DBType.NEO4J).vertex_config if schema else None
        )
        for c in vertex_config.vertex_set:
            index_list = list(
                schema.db_profile.vertex_secondary_indexes(c)
                if schema is not None
                else []
            )
            if db_vertex:
                identity_fields = db_vertex.identity_fields(c)
                if identity_fields:
                    identity_idx = Index(fields=identity_fields, unique=True)
                    seen = {tuple(ix.fields) for ix in index_list}
                    if tuple(identity_idx.fields) not in seen:
                        index_list = [identity_idx, *index_list]
            label = db_vertex.vertex_dbname(c) if db_vertex else c
            identity = tuple(db_vertex.identity_fields(c)) if db_vertex else ()
            for index_obj in index_list:
                self._add_index(
                    label, index_obj, is_identity=tuple(index_obj.fields) == identity
                )

    def define_edge_indexes(self, edges: list[Edge], schema: Schema | None = None):
        """Define indexes for relationship types.

        Creates indexes for each relationship type based on the configuration.

        Args:
            edges: List of edge configurations containing index definitions
        """
        for edge in edges:
            index_list = (
                schema.db_profile.edge_secondary_indexes(edge.edge_id)
                if schema is not None
                else []
            )
            for index_obj in index_list:
                rel_label: str | None = None
                if schema is not None:
                    rel_label = schema.db_profile.edge_relation_name(
                        edge.edge_id, default_relation=edge.relation
                    )
                if rel_label is None:
                    rel_label = edge.relation
                if rel_label is None:
                    continue
                self._add_index(rel_label, index_obj, is_vertex_index=False)

    def _add_index(
        self, obj_name, index: Index, is_vertex_index=True, *, is_identity=False
    ):
        """Add an index or a uniqueness constraint to a label or relationship type.

        ``Index.unique`` emits a uniqueness constraint, which Neo4j backs with
        its own index; otherwise a range index. A constraint the server refuses
        raises, except a composite identity constraint (Community edition may
        refuse it), which falls back to an index with a warning that the
        identity is not enforced as unique.

        Args:
            obj_name: Label or relationship type name
            index: Index configuration to create
            is_vertex_index: If True, create index on nodes, otherwise on relationships
            is_identity: Whether *index* is the vertex identity
        """
        fields_str = ", ".join([f"x.{cypher_name(f)}" for f in index.fields])
        index_name = cypher_name(f"{obj_name}_{'_'.join(index.fields)}")
        label = cypher_name(obj_name)
        formula = f"(x:{label})" if is_vertex_index else f"()-[x:{label}]-()"
        plain = (
            f"CREATE INDEX {index_name} IF NOT EXISTS FOR {formula} ON ({fields_str});"
        )
        if not (index.unique and index.fields):
            _create_schema_rule(self, plain)
            return
        req = fields_str if len(index.fields) == 1 else f"({fields_str})"
        constraint = (
            f"CREATE CONSTRAINT {index_name} IF NOT EXISTS "
            f"FOR {formula} REQUIRE {req} IS UNIQUE;"
        )
        try:
            _create_schema_rule(self, constraint)
        except ClientError as e:
            if not is_identity or len(index.fields) == 1:
                raise
            logger.warning(
                "Neo4j refused a uniqueness constraint on %s(%s): %s. Creating an "
                "index instead; the identity is not enforced as unique.",
                obj_name,
                ", ".join(index.fields),
                e.message,
            )
            _create_schema_rule(self, plain)

    def _drop_all_user_indexes_and_constraints(self) -> None:
        """Remove non-system indexes and constraints for the active database.

        Neo4j keeps indexes when only nodes are deleted; test ``clean_db`` and
        ``recreate_schema`` expect a fully empty catalog. Builtin LOOKUP indexes
        are not dropped.
        """
        try:
            con_cursor = self.execute("SHOW CONSTRAINTS")
            rows = con_cursor.data()
        except ClientError as e:
            logger.warning("Could not list constraints: %s", e)
            rows = []
        for row in rows:
            name = row.get("name")
            if not name:
                continue
            esc = cypher_name(str(name), quoted=True)
            try:
                self.execute(f"DROP CONSTRAINT {esc} IF EXISTS")
            except ClientError as drop_exc:
                logger.debug("DROP CONSTRAINT %s: %s", name, drop_exc)

        try:
            idx_cursor = self.execute("SHOW INDEXES")
            idx_rows = idx_cursor.data()
        except ClientError as e:
            logger.warning("Could not list indexes: %s", e)
            idx_rows = []
        for row in idx_rows:
            name = row.get("name")
            if not name:
                continue
            idx_type = row.get("type") or row.get("indexType") or ""
            if str(idx_type).upper() == "LOOKUP":
                continue
            esc = cypher_name(str(name), quoted=True)
            try:
                self.execute(f"DROP INDEX {esc} IF EXISTS")
            except ClientError as drop_exc:
                logger.debug("DROP INDEX %s: %s", name, drop_exc)

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

        Note: This is a no-op in Neo4j as vertex/edge classes (labels/relationship types) are implicit.
        Field types are still validated (LIST is storable; unsupported types raise).

        Args:
            schema: Schema containing vertex and edge class definitions
        """
        assert_schema_supported(self.flavor, schema)

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

        Note: This is a no-op in Neo4j as vertex classes (labels) are implicit.

        Args:
            schema: Schema containing vertex definitions
        """

    def define_edge_classes(self, edges: list[Edge]):
        """Define edge classes based on schema.

        Note: This is a no-op in Neo4j as edge classes (relationship types) are implicit.

        Args:
            edges: List of edge configurations
        """

    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 (nodes and relationships) from Neo4j.

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

        Args:
            vertex_types: Label names to delete nodes for
            graph_names: Unused in Neo4j (no explicit graph concept)
            delete_all: If True, delete all nodes and relationships
        """
        cnames = vertex_types
        if cnames:
            for c in cnames:
                q = f"MATCH (n:{cypher_name(c)}) DETACH DELETE n"
                self.execute(q)
            return

        if delete_all:
            self._drop_all_user_indexes_and_constraints()
            self.execute("MATCH (n) DETACH DELETE n")
            return

        logger.debug("No vertex_types provided and delete_all=False; no data deleted")

    def _resolve_db_name(self, schema: Schema) -> str:
        return self.config.database or schema.effective_namespace(DBType.NEO4J)

    def _use_database(self, name: str) -> None:
        """Send every later query, of this and of later connections, to *name*."""
        if self.config.database == name:
            return
        self.config.database = name
        self.conn.close()
        self.conn = self._driver.session(database=name)

    def _list_databases(self) -> list[str] | None:
        try:
            result = self.execute("SHOW DATABASES")
            databases: list[str] = []
            for record in result:
                if hasattr(record, "get"):
                    db_name_field = (
                        record.get("name") or record.get("database") or record.get("db")
                    )
                else:
                    db_name_field = getattr(record, "name", None) or getattr(
                        record, "database", None
                    )
                if db_name_field:
                    databases.append(str(db_name_field))
            return databases
        except Exception:
            return None

    def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
        """Ensure the Neo4j database exists, and use it.

        An edition that cannot create it (Community) keeps the home database.
        """
        db_name = self._resolve_db_name(schema)
        databases = self._list_databases()
        if databases is None:
            if create:
                self.create_database(db_name)
            logger.debug(
                "Cannot list Neo4j databases; using the configured or home database"
            )
            return
        if db_name not in databases:
            if not create:
                raise NamespaceNotFoundError(
                    f"Neo4j database '{db_name}' does not exist. "
                    "Create it manually or call with create_namespace=True."
                )
            logger.info(
                "Database '%s' does not exist, attempting to create it...", db_name
            )
            self.create_database(db_name)
            databases = self._list_databases() or []
        if db_name in databases:
            self._use_database(db_name)
        elif self.config.database == db_name:
            raise NamespaceNotFoundError(
                f"Neo4j database '{db_name}' does not exist and the server cannot "
                "create it (Neo4j Community holds one database for your data); "
                "leave the database unset to use the home database"
            )
        else:
            logger.info(
                "Neo4j database '%s' could not be created; using the home database",
                db_name,
            )

    def _node_count(self) -> int:
        result = self.execute("MATCH (n) RETURN count(n) AS c")
        count = 0
        if hasattr(result, "data"):
            data = result.data()
            if data:
                count = data[0].get("c", 0) or 0
        if count == 0 and hasattr(result, "__iter__"):
            for record in result:
                if hasattr(record, "get"):
                    count = record.get("c", 0) or 0
                else:
                    count = getattr(record, "c", 0) or 0
                break
        return int(count)

    def apply_target_schema(
        self,
        schema: Schema,
        *,
        recreate: bool,
        create_namespace: bool = True,
    ) -> None:
        """Define indexes for the schema (labels/relationships are implicit)."""
        assert_schema_supported(self.flavor, schema)
        self.report_edge_direction_support(schema)
        db_name = self._resolve_db_name(schema)
        if self._node_count() > 0 and not recreate:
            raise SchemaExistsError(
                f"Schema/data already exists in database '{db_name}'. "
                "Set recreate_schema=True to replace, or use clear_data=True "
                "before ingestion."
            )
        if recreate:
            try:
                self.delete_database("")
                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_indexes(schema)

    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 the graph without dropping the schema.

        Deletes nodes and relationships for schema-managed labels only.
        """
        vc = schema.resolve_db_aware(DBType.NEO4J).vertex_config
        vertex_types = tuple(vc.vertex_dbname(v) for v in vc.vertex_set)
        if vertex_types:
            self.delete_graph_structure(vertex_types=vertex_types)

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

    def bulk_begin(
        self,
        schema: Schema,
        cfg: BulkConfig,
        *,
        recreate: bool = False,
        resume: bool = False,
    ) -> BulkPlan:
        """Guard the first write; indexes wait until the vertices are written."""
        assert_schema_supported(self.flavor, schema)
        self.report_edge_direction_support(schema)
        self.ensure_target_namespace(schema, create=True)
        if resume:
            return BulkPlan()
        if recreate:
            self.delete_database("")
        elif self._node_count() > 0:
            raise SchemaExistsError(
                f"Schema/data already exists in database "
                f"'{self._resolve_db_name(schema)}'. Set recreate_schema=True to "
                "replace it."
            )
        return BulkPlan()

    def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
        """``CREATE`` the chunk's nodes; a replay ``MERGE``s them on their identity."""
        if chunk.replay:
            self._add_index(
                chunk.class_name,
                Index(fields=chunk.match_keys, unique=True),
                is_identity=True,
            )
            self._await_indexes()
            self.upsert_docs_batch(chunk.rows, chunk.class_name, chunk.match_keys)
            return
        self._write(create_nodes_query(chunk.class_name), chunk.rows)

    def bulk_prepare_edges(self, schema: Schema) -> None:
        """Index every vertex identity, so edges find their endpoints."""
        vc = schema.resolve_db_aware(DBType.NEO4J).vertex_config
        for vertex in vc.vertex_set:
            fields = vc.identity_fields(vertex)
            if fields:
                self._add_index(
                    vc.vertex_dbname(vertex),
                    Index(fields=fields, unique=True),
                    is_identity=True,
                )
        self._await_indexes()

    def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
        """``CREATE`` the chunk's relationships; a replay ``MERGE``s them."""
        relation = chunk.relation_name
        if relation is None:
            raise ValueError(f"edge chunk {chunk.chunk_id} has no relation name")
        if chunk.replay:
            self.insert_edges_batch(
                [list(row) for row in chunk.rows],
                chunk.source_class,
                chunk.target_class,
                relation,
                tuple(chunk.match_keys_source),
                tuple(chunk.match_keys_target),
                relationship_merge_properties=chunk.merge_properties or None,
            )
            return
        query = create_relationships_query(
            chunk.source_class,
            chunk.target_class,
            relation,
            chunk.match_keys_source,
            chunk.match_keys_target,
        )
        self._write(query, [list(row) for row in chunk.rows])

    def bulk_finish(self, schema: Schema) -> BulkCounts:
        """Build the remaining indexes; count what the database holds."""
        self.define_indexes(schema)
        self._await_indexes()
        return bulk_counts(lambda query: _rows(self.execute(query)))

    # -- record sync ---------------------------------------------------------------

    def sync_begin(self, schema: Schema, edge_types: list[SyncEdgeType]) -> None:
        """Index every vertex identity, and each edge's owner property."""
        assert_schema_supported(self.flavor, schema)
        self.bulk_prepare_edges(schema)
        for edge in edge_types:
            if edge.owner_property is not None:
                self._add_index(
                    edge.relation_name,
                    Index(fields=[edge.owner_property]),
                    is_vertex_index=False,
                )
        self._await_indexes()

    def sync_ensure_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
        """``MERGE`` each vertex on its key; only a created one takes the row."""
        self._write(ensure_nodes_query(chunk.class_name, chunk.match_keys), chunk.rows)

    def sync_write_owners(self, schema: Schema, chunk: SyncOwnerChunk) -> SyncWritten:
        """Rewrite the changed owners in one write transaction."""
        read = stored_hashes_query(chunk.class_name, chunk.match_keys, chunk.hash_field)
        stored = {
            row["i"]: row["h"]
            for row in _rows(self.execute(read, rows=hash_rows(chunk)))
        }
        changed = changed_owners(chunk, stored)
        statements = owner_statements(chunk, changed)
        if statements:

            def work(tx: Any) -> None:
                for query, params in statements:
                    tx.run(cast(LiteralString, query), **params).consume()

            self.conn.execute_write(work)
        return SyncWritten(
            written=len(changed), skipped=len(chunk.owners) - len(changed)
        )

    def _write(self, query: str, rows: list[Any]) -> None:
        """Run *query* over ``$rows`` in a write transaction the driver retries.

        A transient error (a lock wait turned deadlock) rolls the transaction
        back, so retrying a ``CREATE`` cannot duplicate it.
        """
        q = cast(LiteralString, query)
        self.conn.execute_write(lambda tx: tx.run(q, rows=rows).consume())

    def _await_indexes(self) -> None:
        """Wait until every index is online; building one may take long."""
        self.execute("CALL db.awaitIndexes(86400)").consume()

    def upsert_docs_batch(self, docs, class_name, match_keys, **kwargs):
        """Upsert a batch of nodes using Cypher.

        Performs an upsert operation on a batch of nodes, using the specified
        match keys to determine whether to update existing nodes or create new ones.

        Args:
            docs: List of node documents to upsert
            class_name: Label to upsert into
            match_keys: Keys to match for upsert operation
            **kwargs: Additional options:
                - dry: If True, don't execute the query
                - union_fields: List properties written as the union of the
                  stored and the given members rather than replaced.
        """
        dry = kwargs.pop("dry", False)
        union_fields: tuple[str, ...] = tuple(kwargs.pop("union_fields", ()) or ())

        params: dict[str, Any]
        if union_fields:
            q = union_upsert_query(class_name, match_keys, union_fields)
            params = {"data": union_upsert_rows(docs, union_fields)}
        else:
            index_str = ", ".join(
                [f"{cypher_name(k)}: row.{cypher_name(k)}" for k in match_keys]
            )
            q = f"""
            WITH $batch AS batch
            UNWIND batch as row
            MERGE (n:{cypher_name(class_name)} {{ {index_str} }})
            ON MATCH set n += row
            ON CREATE set n += row
        """
            params = {"batch": docs}
        if not dry:
            # A managed write transaction, as for edges: a deadlock with a concurrent
            # write is a transient error the driver retries, and a MERGE is safe to
            # repeat. It has committed, or raised, before the call returns.
            query = cast(LiteralString, q)
            self.conn.execute_write(
                lambda tx, query=query, params=params: tx.run(query, **params).consume()
            )

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

        Creates relationships between source and target nodes, with support for
        property matching and unique constraints.

        Args:
            docs_edges: List of edge documents in format [{__source: source_doc, __target: target_doc}]
            source_class: Source node label
            target_class: Target node label
            relation_name: Relationship type name
            match_keys_source: Keys to match source nodes
            match_keys_target: Keys to match target nodes
            filter_uniques: Unused in Neo4j (MERGE handles uniqueness automatically)
            head: Optional limit on number of relationships to insert
            **kwargs: Additional options:
                - dry: If True, don't execute the query
                - collection_name: Unused in Neo4j (kept for interface compatibility)
                - uniq_weight_fields: Unused (ArangoDB upsert); use relationship_merge_properties instead
                - uniq_weight_collections: Unused in Neo4j (ArangoDB-specific)
                - relationship_merge_properties: Property names included in ``MERGE`` so parallel
                  edges (same endpoints and type, different weights) are distinct.
                - union_fields: List properties written as the union of the
                  stored list and the row's.
        """
        opts = consume_insert_edges_kwargs(kwargs)
        dry = opts.dry
        relationship_merge_properties_raw = opts.relationship_merge_properties
        union = opts.union_fields

        # Apply head limit if specified
        if head is not None and isinstance(docs_edges, list):
            docs_edges = docs_edges[:head]

        # Note: filter_uniques is unused because Neo4j's MERGE handles uniqueness automatically

        source_match_str = [
            f"source.{k} = row[0].{k}" for k in map(cypher_name, match_keys_source)
        ]
        target_match_str = [
            f"target.{k} = row[1].{k}" for k in map(cypher_name, match_keys_target)
        ]

        match_clause = "WHERE " + " AND ".join(source_match_str + target_match_str)

        merge_props: tuple[str, ...] = tuple(relationship_merge_properties_raw or ())
        rows = list(docs_edges or [])
        groups = (
            partition_by_absent_merge_props(
                rows,
                merge_props,
                props_of=lambda row: row[2] if len(row) > 2 else None,
                ends_of=lambda row: (row[0], row[1]),
                union=union,
            )
            if merge_props or union
            else [((), rows)]
        )
        for absent, group in groups:
            upsert = rel_upsert_clause(relation_name, merge_props, absent, union=union)
            if union:
                group = rel_union_rows(group, union)
            q = f"""
            WITH $batch AS batch
            UNWIND batch as row
            MATCH (source:{cypher_name(source_class)}),
                  (target:{cypher_name(target_class)}) {match_clause}
            {upsert}
            """
            if not dry:
                # A managed write transaction: edge types sharing endpoint nodes
                # are written concurrently, and the deadlocks that can cause are
                # transient errors the driver retries.
                query = cast(LiteralString, q)
                self.conn.execute_write(
                    lambda tx, query=query, group=group: tx.run(
                        query, batch=group
                    ).consume()
                )

    def insert_return_batch(
        self, docs: list[dict[str, Any]], class_name: str
    ) -> list[dict[str, Any]] | str:
        """Not supported: nodes are written with :meth:`upsert_docs_batch`.

        Raises:
            NotImplementedError: Always.
        """
        raise NotImplementedError(
            f"insert_return_batch is not supported by {type(self).__name__}; "
            "write nodes with upsert_docs_batch and read them back with fetch_docs"
        )

    def fetch_docs(
        self,
        class_name,
        filters: list | dict | None = None,
        limit: int | None = None,
        return_keys: list | None = None,
        unset_keys: list | None = None,
        **kwargs,
    ):
        """Fetch nodes from a label.

        Args:
            class_name: Label to fetch from
            filters: Query filters
            limit: Maximum number of nodes to return
            return_keys: Keys to return
            unset_keys: Unused in Neo4j

        Returns:
            list: Fetched nodes
        """
        params = BoundParams(self.expression_flavor())
        if filters is not None:
            ff = parse_filter_expression(filters)
            filter_clause = f"WHERE {ff(doc_name='n', kind=params.kind, params=params)}"
        else:
            filter_clause = ""

        if return_keys is not None:
            keep_clause_ = ", ".join([f".{cypher_name(item)}" for item in return_keys])
            keep_clause = f"{{ {keep_clause_} }}"
        else:
            keep_clause = ""

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

        q = (
            f"MATCH (n:{cypher_name(class_name)})"
            f"  {filter_clause}"
            f"  RETURN n {keep_clause}"
            f"  {limit_clause}"
        )
        cursor = self.execute(q, **params.values)
        r = [item["n"] for item in _rows(cursor)]
        return r

    # TODO test
    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 | dict | None = None,
        limit: int | None = None,
        return_keys: list | None = None,
        unset_keys: list | None = None,
        direction: EdgeDirection = EdgeDirection.OUT,
        **kwargs,
    ):
        """Fetch edges from Neo4j using Cypher.

        Relationships are stored bidirectionally, so ``IN`` and ``ANY`` are as
        cheap as ``OUT`` — the pattern simply drops or flips its arrow.

        Args:
            from_type: Anchor node label
            from_id: Anchor node ID (property name depends on match_keys used)
            edge_type: Optional relationship type to filter by
            to_type: Optional label 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) - not supported in Neo4j
            direction: Orientations to follow from the anchor
            **kwargs: Additional parameters

        Returns:
            list: List of fetched edges
        """
        # Build Cypher query to fetch edges
        # Match source node first
        # Ids come from callers and travel as parameters, never as query text.
        params: dict[str, Any] = {"from_id": from_id}
        source_match = f"(source:{cypher_name(from_type)} {{id: $from_id}})"
        rel_pattern = cypher_rel_pattern(edge_type, direction)

        # Build target node match
        if to_type:
            target_match = f"(target:{cypher_name(to_type)})"
        else:
            target_match = "(target)"

        # Add target ID filter if provided
        where_clauses: list[str] = []
        if to_id:
            where_clauses.append("target.id = $to_id")
            params["to_id"] = to_id

        # Add additional filters if provided
        if filters is not None:
            bound = BoundParams(self.expression_flavor())
            ff = parse_filter_expression(filters)
            where_clauses.append(
                render_conjunct(ff, doc_name="r", kind=bound.kind, params=bound)
            )
            params.update(bound.values)

        where_clause = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""

        # Build return clause
        if return_keys is not None:
            return_clause = ", ".join(
                [f"r.{k} as {k}" for k in map(cypher_name, return_keys)]
            )
            return_clause = f"RETURN {return_clause}"
        else:
            return_clause = "RETURN r"

        limit_clause = f"LIMIT {limit}" if limit else ""

        query = f"""
            MATCH {source_match}{rel_pattern}{target_match}
            {where_clause}
            {return_clause}
            {limit_clause}
        """

        cursor = self.execute(query, **params)
        result = [item["r"] for item in _rows(cursor)]

        # Note: unset_keys is not supported in Neo4j as we can't modify the result structure
        # after the query

        return result

    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 nodes with ``DETACH DELETE``, which removes their relationships too."""
        delete_nodes(
            lambda query, rows: self.execute(query, data=rows),
            class_name,
            key_docs,
            match_keys,
            chunk_size,
        )

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

    def delete_scoped_vertices(self, class_name: str, field: str, value: Any) -> None:
        """Remove the nodes whose *field* equals *value*, with ``DETACH DELETE``."""
        delete_scoped_nodes(
            lambda query, params: self.execute(query, **params),
            class_name,
            field,
            value,
        )

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

    def fetch_present_documents(
        self,
        batch: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        keep_keys: list[str] | tuple[str, ...] | None = None,
        flatten: bool = False,
        filters: list[Any] | dict[str, Any] | None = None,
    ) -> list[dict[str, Any]] | dict[int, list[dict[str, Any]]]:
        """Stored nodes matching *batch*, by batch position; see :func:`present_documents`.

        With *flatten*, a list of the matches in batch order.
        """
        present = present_documents(
            self, batch, class_name, match_keys, keep_keys=keep_keys, filters=filters
        )
        if flatten:
            return [present[position][0] for position in sorted(present)]
        return present

    def aggregate(
        self,
        class_name,
        aggregation_function: AggregationType,
        discriminant: str | None = None,
        aggregated_field: str | None = None,
        filters: FilterExpression | list | dict | None = None,
    ):
        """Aggregate over nodes of a label.

        Args:
            class_name: Label to aggregate.
            aggregation_function: COUNT, MAX, MIN, AVERAGE or SORTED_UNIQUE.
            discriminant: Field to group by. COUNT only; returns a mapping.
            aggregated_field: Field to aggregate. Required for everything but
                an ungrouped COUNT.
            filters: Applied before aggregating.

        Returns:
            An ungrouped COUNT returns an int; a grouped COUNT returns
            ``{value: count}``; MAX/MIN/AVERAGE return a scalar or ``None``;
            SORTED_UNIQUE returns a list.

        Raises:
            ValueError: If *aggregated_field* is missing where it is required,
                or the aggregation type is not supported.
        """
        filter_clause = ""
        params = BoundParams(self.expression_flavor())
        if filters is not None:
            expression = parse_filter_expression(filters)
            filter_clause = (
                f" WHERE {expression(doc_name='n', kind=params.kind, params=params)}"
            )

        label = cypher_name(str(class_name), quoted=True)
        query = f"MATCH (n:{label}){filter_clause}"

        if aggregation_function == AggregationType.COUNT:
            if discriminant:
                field = cypher_name(discriminant, quoted=True)
                query += f" RETURN n.{field} AS key, count(*) AS value"
                rows = _rows(self.execute(query, **params.values))
                return {row["key"]: row["value"] for row in rows}
            query += " RETURN count(n) AS value"
        else:
            if not aggregated_field:
                raise ValueError(
                    f"aggregated_field is required for {aggregation_function}"
                )
            field = cypher_name(aggregated_field, quoted=True)
            if aggregation_function == AggregationType.MAX:
                query += f" RETURN max(n.{field}) AS value"
            elif aggregation_function == AggregationType.MIN:
                query += f" RETURN min(n.{field}) AS value"
            elif aggregation_function == AggregationType.AVERAGE:
                query += f" RETURN avg(n.{field}) AS value"
            elif aggregation_function == AggregationType.SORTED_UNIQUE:
                query += f" RETURN DISTINCT n.{field} AS value ORDER BY n.{field}"
            else:
                raise ValueError(
                    f"Unsupported aggregation type: {aggregation_function}"
                )

        rows = _rows(self.execute(query, **params.values))
        if aggregation_function == AggregationType.SORTED_UNIQUE:
            return [row["value"] for row in rows]
        return rows[0]["value"] if rows else None

    def keep_absent_documents(
        self,
        batch: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        keep_keys: list[str] | tuple[str, ...] | None = None,
        filters: list[Any] | dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        """The documents of *batch* with no stored node; see :func:`absent_documents`."""
        return absent_documents(
            self, batch, class_name, match_keys, keep_keys=keep_keys, filters=filters
        )

    def introspect_graph_schema(
        self,
        schema_name: str | None = None,
        *,
        sample_limit: int = 100,
    ) -> Schema:
        """Infer a graflo Schema from Neo4j via label and relationship sampling."""
        from graflo.db.cypher.introspection import collect_cypher_introspection

        resolved_name = schema_name or self.config.database or "neo4j"

        def run(query: str, keys: Sequence[str]) -> list[dict[str, Any]]:
            return self.execute(query).data()

        introspection = collect_cypher_introspection(
            name=resolved_name, run=run, sample_limit=sample_limit
        )
        return GraphSchemaInferencer(db_flavor=DBType.NEO4J).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,
        guard: FilterExpression | None = None,
    ) -> GraphContainer:
        """Bounded neighbourhood via a single variable-length Cypher pattern.

        Native rather than composed from :meth:`fetch_edges`, and not only for
        speed: ``fetch_edges`` here returns ``RETURN r``, and the driver renders
        a bare relationship without its endpoints, so a breadth-first walk over
        it has nothing to step to.
        """
        return cypher_graph_neighbors(
            self,
            vertex_type=vertex_type,
            key=key,
            hops=hops,
            direction=direction,
            edge_types=edge_types,
            limit=limit,
            schema=schema,
            run=lambda query, params: _rows(self.execute(query, **params)),
            filters=filters,
            guard=guard,
        )

    def fetch_all_docs(
        self,
        class_name: str,
        *,
        limit: int | None = None,
    ) -> list[dict[str, Any]]:
        """Fetch all nodes for a label."""
        escaped = cypher_name(class_name, quoted=True)
        limit_clause = f"LIMIT {limit}" if limit is not None else ""
        query = f"""
            MATCH (n:{escaped})
            RETURN properties(n) AS props
            {limit_clause}
        """
        rows = _rows(self.execute(query))
        return [
            strip_internal_properties(row["props"]) for row in rows if row.get("props")
        ]

    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 relationships between two labels."""
        src_esc = cypher_name(source_class, quoted=True)
        tgt_esc = cypher_name(target_class, quoted=True)
        rel_clause = ""
        if relation_name:
            rel_clause = f":{cypher_name(relation_name, quoted=True)}"
        limit_clause = f"LIMIT {limit}" if limit is not None else ""
        query = f"""
            MATCH (source:{src_esc})-[r{rel_clause}]->(target:{tgt_esc})
            RETURN properties(source) AS source_props,
                   properties(target) AS target_props,
                   properties(r) AS edge_props
            {limit_clause}
        """
        rows = _rows(self.execute(query))
        result: list[list[dict[str, Any]]] = []
        for row in rows:
            source_doc = strip_internal_properties(row.get("source_props") or {})
            target_doc = strip_internal_properties(row.get("target_props") or {})
            edge_props = strip_internal_properties(row.get("edge_props") or {})
            result.append([source_doc, target_doc, edge_props])
        return result

Attributes

bulk_partition_edges = True class-attribute instance-attribute

config = config instance-attribute

conn = self._driver.session(database=config.database) instance-attribute

flavor = DBType.NEO4J 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_membership_read = True class-attribute instance-attribute

supports_read_guard = True class-attribute instance-attribute

supports_schema_ddl = True class-attribute instance-attribute

supports_schema_introspection = True class-attribute instance-attribute

supports_scoped_delete = True class-attribute instance-attribute

supports_sync = True class-attribute instance-attribute

supports_union_merge = True class-attribute instance-attribute

Methods:

__init__(config)

Initialize Neo4j connection.

Parameters:

Name	Type	Description	Default
config	Neo4jConfig	Neo4j connection configuration containing URL and credentials	required

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

def __init__(self, config: Neo4jConfig):
    """Initialize Neo4j connection.

    Args:
        config: Neo4j connection configuration containing URL and credentials
    """
    super().__init__()
    # Store config for later use
    self.config = config
    # Ensure url is not None - GraphDatabase.driver requires a non-None URI
    if config.url is None:
        raise ValueError("Neo4j connection requires a URL to be configured")
    # A password is never dropped: without a username it is Neo4j's built-in
    # user's, and only a config with no password connects unauthenticated.
    auth = None
    if config.password is not None:
        auth = (config.username or DEFAULT_NEO4J_USERNAME, config.password)
    # Queries are built from the schema, so an unrecognized label, property or
    # relation is one nothing has written yet, never a typo: no notification.
    self._driver = GraphDatabase.driver(
        uri=config.url,
        auth=auth,
        notifications_disabled_categories=[
            NotificationDisabledCategory.UNRECOGNIZED
        ],
    )
    # ``None`` is the user's home database.
    self.conn = self._driver.session(database=config.database)

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

Aggregate over nodes of a label.

Parameters:

Name	Type	Description	Default
class_name		Label to aggregate.	required
aggregation_function	AggregationType	COUNT, MAX, MIN, AVERAGE or SORTED_UNIQUE.	required
discriminant	str | None	Field to group by. COUNT only; returns a mapping.	None
aggregated_field	str | None	Field to aggregate. Required for everything but an ungrouped COUNT.	None
filters	FilterExpression | list | dict | None	Applied before aggregating.	None

Returns:

Type	Description
	An ungrouped COUNT returns an int; a grouped COUNT returns
	{value: count}; MAX/MIN/AVERAGE return a scalar or None;
	SORTED_UNIQUE returns a list.

Raises:

Type	Description
ValueError	If aggregated_field is missing where it is required, or the aggregation type is not supported.

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

def aggregate(
    self,
    class_name,
    aggregation_function: AggregationType,
    discriminant: str | None = None,
    aggregated_field: str | None = None,
    filters: FilterExpression | list | dict | None = None,
):
    """Aggregate over nodes of a label.

    Args:
        class_name: Label to aggregate.
        aggregation_function: COUNT, MAX, MIN, AVERAGE or SORTED_UNIQUE.
        discriminant: Field to group by. COUNT only; returns a mapping.
        aggregated_field: Field to aggregate. Required for everything but
            an ungrouped COUNT.
        filters: Applied before aggregating.

    Returns:
        An ungrouped COUNT returns an int; a grouped COUNT returns
        ``{value: count}``; MAX/MIN/AVERAGE return a scalar or ``None``;
        SORTED_UNIQUE returns a list.

    Raises:
        ValueError: If *aggregated_field* is missing where it is required,
            or the aggregation type is not supported.
    """
    filter_clause = ""
    params = BoundParams(self.expression_flavor())
    if filters is not None:
        expression = parse_filter_expression(filters)
        filter_clause = (
            f" WHERE {expression(doc_name='n', kind=params.kind, params=params)}"
        )

    label = cypher_name(str(class_name), quoted=True)
    query = f"MATCH (n:{label}){filter_clause}"

    if aggregation_function == AggregationType.COUNT:
        if discriminant:
            field = cypher_name(discriminant, quoted=True)
            query += f" RETURN n.{field} AS key, count(*) AS value"
            rows = _rows(self.execute(query, **params.values))
            return {row["key"]: row["value"] for row in rows}
        query += " RETURN count(n) AS value"
    else:
        if not aggregated_field:
            raise ValueError(
                f"aggregated_field is required for {aggregation_function}"
            )
        field = cypher_name(aggregated_field, quoted=True)
        if aggregation_function == AggregationType.MAX:
            query += f" RETURN max(n.{field}) AS value"
        elif aggregation_function == AggregationType.MIN:
            query += f" RETURN min(n.{field}) AS value"
        elif aggregation_function == AggregationType.AVERAGE:
            query += f" RETURN avg(n.{field}) AS value"
        elif aggregation_function == AggregationType.SORTED_UNIQUE:
            query += f" RETURN DISTINCT n.{field} AS value ORDER BY n.{field}"
        else:
            raise ValueError(
                f"Unsupported aggregation type: {aggregation_function}"
            )

    rows = _rows(self.execute(query, **params.values))
    if aggregation_function == AggregationType.SORTED_UNIQUE:
        return [row["value"] for row in rows]
    return rows[0]["value"] if rows else None

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

Define indexes for the schema (labels/relationships are implicit).

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

def apply_target_schema(
    self,
    schema: Schema,
    *,
    recreate: bool,
    create_namespace: bool = True,
) -> None:
    """Define indexes for the schema (labels/relationships are implicit)."""
    assert_schema_supported(self.flavor, schema)
    self.report_edge_direction_support(schema)
    db_name = self._resolve_db_name(schema)
    if self._node_count() > 0 and not recreate:
        raise SchemaExistsError(
            f"Schema/data already exists in database '{db_name}'. "
            "Set recreate_schema=True to replace, or use clear_data=True "
            "before ingestion."
        )
    if recreate:
        try:
            self.delete_database("")
            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_indexes(schema)

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

Guard the first write; indexes wait until the vertices are written.

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

def bulk_begin(
    self,
    schema: Schema,
    cfg: BulkConfig,
    *,
    recreate: bool = False,
    resume: bool = False,
) -> BulkPlan:
    """Guard the first write; indexes wait until the vertices are written."""
    assert_schema_supported(self.flavor, schema)
    self.report_edge_direction_support(schema)
    self.ensure_target_namespace(schema, create=True)
    if resume:
        return BulkPlan()
    if recreate:
        self.delete_database("")
    elif self._node_count() > 0:
        raise SchemaExistsError(
            f"Schema/data already exists in database "
            f"'{self._resolve_db_name(schema)}'. Set recreate_schema=True to "
            "replace it."
        )
    return BulkPlan()

bulk_finish(schema)

Build the remaining indexes; count what the database holds.

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

def bulk_finish(self, schema: Schema) -> BulkCounts:
    """Build the remaining indexes; count what the database holds."""
    self.define_indexes(schema)
    self._await_indexes()
    return bulk_counts(lambda query: _rows(self.execute(query)))

bulk_prepare_edges(schema)

Index every vertex identity, so edges find their endpoints.

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

def bulk_prepare_edges(self, schema: Schema) -> None:
    """Index every vertex identity, so edges find their endpoints."""
    vc = schema.resolve_db_aware(DBType.NEO4J).vertex_config
    for vertex in vc.vertex_set:
        fields = vc.identity_fields(vertex)
        if fields:
            self._add_index(
                vc.vertex_dbname(vertex),
                Index(fields=fields, unique=True),
                is_identity=True,
            )
    self._await_indexes()

bulk_write_edges(schema, chunk)

CREATE the chunk's relationships; a replay MERGEs them.

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

def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
    """``CREATE`` the chunk's relationships; a replay ``MERGE``s them."""
    relation = chunk.relation_name
    if relation is None:
        raise ValueError(f"edge chunk {chunk.chunk_id} has no relation name")
    if chunk.replay:
        self.insert_edges_batch(
            [list(row) for row in chunk.rows],
            chunk.source_class,
            chunk.target_class,
            relation,
            tuple(chunk.match_keys_source),
            tuple(chunk.match_keys_target),
            relationship_merge_properties=chunk.merge_properties or None,
        )
        return
    query = create_relationships_query(
        chunk.source_class,
        chunk.target_class,
        relation,
        chunk.match_keys_source,
        chunk.match_keys_target,
    )
    self._write(query, [list(row) for row in chunk.rows])

bulk_write_vertices(schema, chunk)

CREATE the chunk's nodes; a replay MERGEs them on their identity.

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

def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
    """``CREATE`` the chunk's nodes; a replay ``MERGE``s them on their identity."""
    if chunk.replay:
        self._add_index(
            chunk.class_name,
            Index(fields=chunk.match_keys, unique=True),
            is_identity=True,
        )
        self._await_indexes()
        self.upsert_docs_batch(chunk.rows, chunk.class_name, chunk.match_keys)
        return
    self._write(create_nodes_query(chunk.class_name), chunk.rows)

clear_data(schema)

Remove all data from the graph without dropping the schema.

Deletes nodes and relationships for schema-managed labels only.

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

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

    Deletes nodes and relationships for schema-managed labels only.
    """
    vc = schema.resolve_db_aware(DBType.NEO4J).vertex_config
    vertex_types = tuple(vc.vertex_dbname(v) for v in vc.vertex_set)
    if vertex_types:
        self.delete_graph_structure(vertex_types=vertex_types)

close()

Close the Neo4j connection and session.

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

def close(self):
    """Close the Neo4j connection and session."""
    # Close session first, then the underlying driver
    try:
        self.conn.close()
    finally:
        # Ensure the driver is also closed to release resources
        self._driver.close()

create_database(name)

Create a new Neo4j database.

Note: This operation is only supported in Neo4j Enterprise Edition. Community Edition only supports one database per instance.

Parameters:

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

Raises:

Type	Description
Exception	If database creation fails and it's not a Community Edition limitation

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

def create_database(self, name: str):
    """Create a new Neo4j database.

    Note: This operation is only supported in Neo4j Enterprise Edition.
    Community Edition only supports one database per instance.

    Args:
        name: Name of the database to create

    Raises:
        Exception: If database creation fails and it's not a Community Edition limitation
    """
    try:
        self.execute(f"CREATE DATABASE {cypher_name(name, quoted=True)}")
        logger.info(f"Successfully created Neo4j database '{name}'")
    except Exception as e:
        # Check if this is a Neo4j Community Edition limitation
        error_str = str(e).lower()
        if (
            "unsupported administration command" in error_str
            or "create database" in error_str
        ):
            # This is likely Neo4j Community Edition - don't raise, just log
            logger.info(
                f"Neo4j Community Edition detected: Cannot create database '{name}'. "
                f"Community Edition only supports one database per instance. "
                f"Continuing with default database."
            )
            # Don't raise - allow operation to continue with default database
            return
        # For other errors, raise them
        raise

define_edge_classes(edges)

Define edge classes based on schema.

Note: This is a no-op in Neo4j as edge classes (relationship types) are implicit.

Parameters:

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

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

def define_edge_classes(self, edges: list[Edge]):
    """Define edge classes based on schema.

    Note: This is a no-op in Neo4j as edge classes (relationship types) are implicit.

    Args:
        edges: List of edge configurations
    """

define_edge_indexes(edges, schema=None)

Define indexes for relationship types.

Creates indexes for each relationship type 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/neo4j/conn.py

def define_edge_indexes(self, edges: list[Edge], schema: Schema | None = None):
    """Define indexes for relationship types.

    Creates indexes for each relationship type based on the configuration.

    Args:
        edges: List of edge configurations containing index definitions
    """
    for edge in edges:
        index_list = (
            schema.db_profile.edge_secondary_indexes(edge.edge_id)
            if schema is not None
            else []
        )
        for index_obj in index_list:
            rel_label: str | None = None
            if schema is not None:
                rel_label = schema.db_profile.edge_relation_name(
                    edge.edge_id, default_relation=edge.relation
                )
            if rel_label is None:
                rel_label = edge.relation
            if rel_label is None:
                continue
            self._add_index(rel_label, index_obj, is_vertex_index=False)

define_schema(schema)

Define vertex and edge classes based on schema.

Note: This is a no-op in Neo4j as vertex/edge classes (labels/relationship types) are implicit. Field types are still validated (LIST is storable; unsupported types raise).

Parameters:

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

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

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

    Note: This is a no-op in Neo4j as vertex/edge classes (labels/relationship types) are implicit.
    Field types are still validated (LIST is storable; unsupported types raise).

    Args:
        schema: Schema containing vertex and edge class definitions
    """
    assert_schema_supported(self.flavor, schema)

define_vertex_classes(schema)

Define vertex classes based on schema.

Note: This is a no-op in Neo4j as vertex classes (labels) are implicit.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema containing vertex definitions	required

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

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

    Note: This is a no-op in Neo4j as vertex classes (labels) are implicit.

    Args:
        schema: Schema containing vertex definitions
    """

define_vertex_indexes(vertex_config, schema=None)

Define indexes for vertex labels.

Creates indexes for each vertex label based on the configuration. Neo4j has no implicit identity index; the identity, as a uniqueness constraint, is prepended when schema is provided.

Parameters:

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

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

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

    Creates indexes for each vertex label based on the configuration.
    Neo4j has no implicit identity index; the identity, as a uniqueness
    constraint, is prepended when schema is provided.

    Args:
        vertex_config: Vertex configuration containing index definitions
    """
    if schema is None:
        logger.warning(
            "Schema is None: identity indexes cannot be ensured without schema"
        )
    db_vertex = (
        schema.resolve_db_aware(DBType.NEO4J).vertex_config if schema else None
    )
    for c in vertex_config.vertex_set:
        index_list = list(
            schema.db_profile.vertex_secondary_indexes(c)
            if schema is not None
            else []
        )
        if db_vertex:
            identity_fields = db_vertex.identity_fields(c)
            if identity_fields:
                identity_idx = Index(fields=identity_fields, unique=True)
                seen = {tuple(ix.fields) for ix in index_list}
                if tuple(identity_idx.fields) not in seen:
                    index_list = [identity_idx, *index_list]
        label = db_vertex.vertex_dbname(c) if db_vertex else c
        identity = tuple(db_vertex.identity_fields(c)) if db_vertex else ()
        for index_obj in index_list:
            self._add_index(
                label, index_obj, is_identity=tuple(index_obj.fields) == identity
            )

delete_database(name)

Delete a Neo4j database.

Note: This operation is only supported in Neo4j Enterprise Edition. As a fallback, it deletes all nodes and relationships.

Parameters:

Name	Type	Description	Default
name	str	Name of the database to delete (unused, deletes all data)	required

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

def delete_database(self, name: str):
    """Delete a Neo4j database.

    Note: This operation is only supported in Neo4j Enterprise Edition.
    As a fallback, it deletes all nodes and relationships.

    Args:
        name: Name of the database to delete (unused, deletes all data)
    """
    try:
        self._drop_all_user_indexes_and_constraints()
        self.execute("MATCH (n) DETACH DELETE n")
        logger.info("Successfully cleaned Neo4j database")
    except Exception as e:
        logger.error(
            f"Failed to clean Neo4j database: {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 the relationships of one type between the given node pairs.

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

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

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

Delete graph structure (nodes and relationships) from Neo4j.

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

Parameters:

Name	Type	Description	Default
vertex_types	tuple[str, ...] | list[str]	Label names to delete nodes for	()
graph_names	tuple[str, ...] | list[str]	Unused in Neo4j (no explicit graph concept)	()
delete_all	bool	If True, delete all nodes and relationships	False

Source code in graflo/db/neo4j/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 (nodes and relationships) from Neo4j.

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

    Args:
        vertex_types: Label names to delete nodes for
        graph_names: Unused in Neo4j (no explicit graph concept)
        delete_all: If True, delete all nodes and relationships
    """
    cnames = vertex_types
    if cnames:
        for c in cnames:
            q = f"MATCH (n:{cypher_name(c)}) DETACH DELETE n"
            self.execute(q)
        return

    if delete_all:
        self._drop_all_user_indexes_and_constraints()
        self.execute("MATCH (n) DETACH DELETE n")
        return

    logger.debug("No vertex_types provided and delete_all=False; no data deleted")

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

Remove the relationships of one type whose field equals value.

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

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

delete_scoped_vertices(class_name, field, value)

Remove the nodes whose field equals value, with DETACH DELETE.

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

def delete_scoped_vertices(self, class_name: str, field: str, value: Any) -> None:
    """Remove the nodes whose *field* equals *value*, with ``DETACH DELETE``."""
    delete_scoped_nodes(
        lambda query, params: self.execute(query, **params),
        class_name,
        field,
        value,
    )

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

Remove nodes with DETACH DELETE, which removes their relationships too.

Source code in graflo/db/neo4j/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 nodes with ``DETACH DELETE``, which removes their relationships too."""
    delete_nodes(
        lambda query, rows: self.execute(query, data=rows),
        class_name,
        key_docs,
        match_keys,
        chunk_size,
    )

ensure_target_namespace(schema, *, create)

Ensure the Neo4j database exists, and use it.

An edition that cannot create it (Community) keeps the home database.

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

def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
    """Ensure the Neo4j database exists, and use it.

    An edition that cannot create it (Community) keeps the home database.
    """
    db_name = self._resolve_db_name(schema)
    databases = self._list_databases()
    if databases is None:
        if create:
            self.create_database(db_name)
        logger.debug(
            "Cannot list Neo4j databases; using the configured or home database"
        )
        return
    if db_name not in databases:
        if not create:
            raise NamespaceNotFoundError(
                f"Neo4j database '{db_name}' does not exist. "
                "Create it manually or call with create_namespace=True."
            )
        logger.info(
            "Database '%s' does not exist, attempting to create it...", db_name
        )
        self.create_database(db_name)
        databases = self._list_databases() or []
    if db_name in databases:
        self._use_database(db_name)
    elif self.config.database == db_name:
        raise NamespaceNotFoundError(
            f"Neo4j database '{db_name}' does not exist and the server cannot "
            "create it (Neo4j Community holds one database for your data); "
            "leave the database unset to use the home database"
        )
    else:
        logger.info(
            "Neo4j database '%s' could not be created; using the home database",
            db_name,
        )

execute(query, **kwargs)

Execute a Cypher query.

Parameters:

Name	Type	Description	Default
query		Cypher query string to execute	required
**kwargs		Additional query parameters	{}

Returns:

Name	Type	Description
Result		Neo4j query result

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

def execute(self, query, **kwargs):
    """Execute a Cypher query.

    Args:
        query: Cypher query string to execute
        **kwargs: Additional query parameters

    Returns:
        Result: Neo4j query result
    """
    cursor = self.conn.run(query, **kwargs)
    return cursor

fetch_all_docs(class_name, *, limit=None)

Fetch all nodes for a label.

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

def fetch_all_docs(
    self,
    class_name: str,
    *,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """Fetch all nodes for a label."""
    escaped = cypher_name(class_name, quoted=True)
    limit_clause = f"LIMIT {limit}" if limit is not None else ""
    query = f"""
        MATCH (n:{escaped})
        RETURN properties(n) AS props
        {limit_clause}
    """
    rows = _rows(self.execute(query))
    return [
        strip_internal_properties(row["props"]) for row in rows if row.get("props")
    ]

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

Fetch all relationships between two labels.

Source code in graflo/db/neo4j/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 relationships between two labels."""
    src_esc = cypher_name(source_class, quoted=True)
    tgt_esc = cypher_name(target_class, quoted=True)
    rel_clause = ""
    if relation_name:
        rel_clause = f":{cypher_name(relation_name, quoted=True)}"
    limit_clause = f"LIMIT {limit}" if limit is not None else ""
    query = f"""
        MATCH (source:{src_esc})-[r{rel_clause}]->(target:{tgt_esc})
        RETURN properties(source) AS source_props,
               properties(target) AS target_props,
               properties(r) AS edge_props
        {limit_clause}
    """
    rows = _rows(self.execute(query))
    result: list[list[dict[str, Any]]] = []
    for row in rows:
        source_doc = strip_internal_properties(row.get("source_props") or {})
        target_doc = strip_internal_properties(row.get("target_props") or {})
        edge_props = strip_internal_properties(row.get("edge_props") or {})
        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 nodes from a label.

Parameters:

Name	Type	Description	Default
class_name		Label to fetch from	required
filters	list | dict | None	Query filters	None
limit	int | None	Maximum number of nodes to return	None
return_keys	list | None	Keys to return	None
unset_keys	list | None	Unused in Neo4j	None

Returns:

Name	Type	Description
list		Fetched nodes

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

def fetch_docs(
    self,
    class_name,
    filters: list | dict | None = None,
    limit: int | None = None,
    return_keys: list | None = None,
    unset_keys: list | None = None,
    **kwargs,
):
    """Fetch nodes from a label.

    Args:
        class_name: Label to fetch from
        filters: Query filters
        limit: Maximum number of nodes to return
        return_keys: Keys to return
        unset_keys: Unused in Neo4j

    Returns:
        list: Fetched nodes
    """
    params = BoundParams(self.expression_flavor())
    if filters is not None:
        ff = parse_filter_expression(filters)
        filter_clause = f"WHERE {ff(doc_name='n', kind=params.kind, params=params)}"
    else:
        filter_clause = ""

    if return_keys is not None:
        keep_clause_ = ", ".join([f".{cypher_name(item)}" for item in return_keys])
        keep_clause = f"{{ {keep_clause_} }}"
    else:
        keep_clause = ""

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

    q = (
        f"MATCH (n:{cypher_name(class_name)})"
        f"  {filter_clause}"
        f"  RETURN n {keep_clause}"
        f"  {limit_clause}"
    )
    cursor = self.execute(q, **params.values)
    r = [item["n"] for item in _rows(cursor)]
    return r

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 Neo4j using Cypher.

Relationships are stored bidirectionally, so IN and ANY are as cheap as OUT — the pattern simply drops or flips its arrow.

Parameters:

Name	Type	Description	Default
from_type	str	Anchor node label	required
from_id	str	Anchor node ID (property name depends on match_keys used)	required
edge_type	str | None	Optional relationship type to filter by	None
to_type	str | None	Optional label of the other endpoint	None
to_id	str | None	Optional ID of the other endpoint	None
filters	list | dict | None	Additional query filters	None
limit	int | None	Maximum number of edges to return	None
return_keys	list | None	Keys to return (projection)	None
unset_keys	list | None	Keys to exclude (projection) - not supported in Neo4j	None
direction	EdgeDirection	Orientations to follow from the anchor	OUT
**kwargs		Additional parameters	{}

Returns:

Name	Type	Description
list		List of fetched edges

Source code in graflo/db/neo4j/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 | dict | None = None,
    limit: int | None = None,
    return_keys: list | None = None,
    unset_keys: list | None = None,
    direction: EdgeDirection = EdgeDirection.OUT,
    **kwargs,
):
    """Fetch edges from Neo4j using Cypher.

    Relationships are stored bidirectionally, so ``IN`` and ``ANY`` are as
    cheap as ``OUT`` — the pattern simply drops or flips its arrow.

    Args:
        from_type: Anchor node label
        from_id: Anchor node ID (property name depends on match_keys used)
        edge_type: Optional relationship type to filter by
        to_type: Optional label 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) - not supported in Neo4j
        direction: Orientations to follow from the anchor
        **kwargs: Additional parameters

    Returns:
        list: List of fetched edges
    """
    # Build Cypher query to fetch edges
    # Match source node first
    # Ids come from callers and travel as parameters, never as query text.
    params: dict[str, Any] = {"from_id": from_id}
    source_match = f"(source:{cypher_name(from_type)} {{id: $from_id}})"
    rel_pattern = cypher_rel_pattern(edge_type, direction)

    # Build target node match
    if to_type:
        target_match = f"(target:{cypher_name(to_type)})"
    else:
        target_match = "(target)"

    # Add target ID filter if provided
    where_clauses: list[str] = []
    if to_id:
        where_clauses.append("target.id = $to_id")
        params["to_id"] = to_id

    # Add additional filters if provided
    if filters is not None:
        bound = BoundParams(self.expression_flavor())
        ff = parse_filter_expression(filters)
        where_clauses.append(
            render_conjunct(ff, doc_name="r", kind=bound.kind, params=bound)
        )
        params.update(bound.values)

    where_clause = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""

    # Build return clause
    if return_keys is not None:
        return_clause = ", ".join(
            [f"r.{k} as {k}" for k in map(cypher_name, return_keys)]
        )
        return_clause = f"RETURN {return_clause}"
    else:
        return_clause = "RETURN r"

    limit_clause = f"LIMIT {limit}" if limit else ""

    query = f"""
        MATCH {source_match}{rel_pattern}{target_match}
        {where_clause}
        {return_clause}
        {limit_clause}
    """

    cursor = self.execute(query, **params)
    result = [item["r"] for item in _rows(cursor)]

    # Note: unset_keys is not supported in Neo4j as we can't modify the result structure
    # after the query

    return result

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

Stored nodes matching batch, by batch position; see :func:present_documents.

With flatten, a list of the matches in batch order.

Source code in graflo/db/neo4j/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: list[Any] | dict[str, Any] | None = None,
) -> list[dict[str, Any]] | dict[int, list[dict[str, Any]]]:
    """Stored nodes matching *batch*, by batch position; see :func:`present_documents`.

    With *flatten*, a list of the matches in batch order.
    """
    present = present_documents(
        self, batch, class_name, match_keys, keep_keys=keep_keys, filters=filters
    )
    if flatten:
        return [present[position][0] for position in sorted(present)]
    return present

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

Convenience wrapper: ensure database then apply schema.

Source code in graflo/db/neo4j/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, match_keys_target, filter_uniques=True, head=None, **kwargs)

Insert a batch of relationships using Cypher.

Creates relationships between source and target nodes, with support for property matching and unique constraints.

Parameters:

Name	Type	Description	Default
docs_edges	list[list[dict[str, Any]]] | list[Any] | None	List of edge documents in format [{__source: source_doc, __target: target_doc}]	required
source_class	str	Source node label	required
target_class	str	Target node label	required
relation_name	str	Relationship type name	required
match_keys_source	tuple[str, ...]	Keys to match source nodes	required
match_keys_target	tuple[str, ...]	Keys to match target nodes	required
filter_uniques	bool	Unused in Neo4j (MERGE handles uniqueness automatically)	True
head	int | None	Optional limit on number of relationships to insert	None
**kwargs	Any	Additional options: - dry: If True, don't execute the query - collection_name: Unused in Neo4j (kept for interface compatibility) - uniq_weight_fields: Unused (ArangoDB upsert); use relationship_merge_properties instead - uniq_weight_collections: Unused in Neo4j (ArangoDB-specific) - relationship_merge_properties: Property names included in MERGE so parallel edges (same endpoints and type, different weights) are distinct. - union_fields: List properties written as the union of the stored list and the row's.	{}

Source code in graflo/db/neo4j/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, ...],
    match_keys_target: tuple[str, ...],
    filter_uniques: bool = True,
    head: int | None = None,
    **kwargs: Any,
) -> None:
    """Insert a batch of relationships using Cypher.

    Creates relationships between source and target nodes, with support for
    property matching and unique constraints.

    Args:
        docs_edges: List of edge documents in format [{__source: source_doc, __target: target_doc}]
        source_class: Source node label
        target_class: Target node label
        relation_name: Relationship type name
        match_keys_source: Keys to match source nodes
        match_keys_target: Keys to match target nodes
        filter_uniques: Unused in Neo4j (MERGE handles uniqueness automatically)
        head: Optional limit on number of relationships to insert
        **kwargs: Additional options:
            - dry: If True, don't execute the query
            - collection_name: Unused in Neo4j (kept for interface compatibility)
            - uniq_weight_fields: Unused (ArangoDB upsert); use relationship_merge_properties instead
            - uniq_weight_collections: Unused in Neo4j (ArangoDB-specific)
            - relationship_merge_properties: Property names included in ``MERGE`` so parallel
              edges (same endpoints and type, different weights) are distinct.
            - union_fields: List properties written as the union of the
              stored list and the row's.
    """
    opts = consume_insert_edges_kwargs(kwargs)
    dry = opts.dry
    relationship_merge_properties_raw = opts.relationship_merge_properties
    union = opts.union_fields

    # Apply head limit if specified
    if head is not None and isinstance(docs_edges, list):
        docs_edges = docs_edges[:head]

    # Note: filter_uniques is unused because Neo4j's MERGE handles uniqueness automatically

    source_match_str = [
        f"source.{k} = row[0].{k}" for k in map(cypher_name, match_keys_source)
    ]
    target_match_str = [
        f"target.{k} = row[1].{k}" for k in map(cypher_name, match_keys_target)
    ]

    match_clause = "WHERE " + " AND ".join(source_match_str + target_match_str)

    merge_props: tuple[str, ...] = tuple(relationship_merge_properties_raw or ())
    rows = list(docs_edges or [])
    groups = (
        partition_by_absent_merge_props(
            rows,
            merge_props,
            props_of=lambda row: row[2] if len(row) > 2 else None,
            ends_of=lambda row: (row[0], row[1]),
            union=union,
        )
        if merge_props or union
        else [((), rows)]
    )
    for absent, group in groups:
        upsert = rel_upsert_clause(relation_name, merge_props, absent, union=union)
        if union:
            group = rel_union_rows(group, union)
        q = f"""
        WITH $batch AS batch
        UNWIND batch as row
        MATCH (source:{cypher_name(source_class)}),
              (target:{cypher_name(target_class)}) {match_clause}
        {upsert}
        """
        if not dry:
            # A managed write transaction: edge types sharing endpoint nodes
            # are written concurrently, and the deadlocks that can cause are
            # transient errors the driver retries.
            query = cast(LiteralString, q)
            self.conn.execute_write(
                lambda tx, query=query, group=group: tx.run(
                    query, batch=group
                ).consume()
            )

insert_return_batch(docs, class_name)

Not supported: nodes are written with :meth:upsert_docs_batch.

Raises:

Type	Description
NotImplementedError	Always.

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

def insert_return_batch(
    self, docs: list[dict[str, Any]], class_name: str
) -> list[dict[str, Any]] | str:
    """Not supported: nodes are written with :meth:`upsert_docs_batch`.

    Raises:
        NotImplementedError: Always.
    """
    raise NotImplementedError(
        f"insert_return_batch is not supported by {type(self).__name__}; "
        "write nodes with upsert_docs_batch and read them back with fetch_docs"
    )

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

Infer a graflo Schema from Neo4j via label and relationship sampling.

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

def introspect_graph_schema(
    self,
    schema_name: str | None = None,
    *,
    sample_limit: int = 100,
) -> Schema:
    """Infer a graflo Schema from Neo4j via label and relationship sampling."""
    from graflo.db.cypher.introspection import collect_cypher_introspection

    resolved_name = schema_name or self.config.database or "neo4j"

    def run(query: str, keys: Sequence[str]) -> list[dict[str, Any]]:
        return self.execute(query).data()

    introspection = collect_cypher_introspection(
        name=resolved_name, run=run, sample_limit=sample_limit
    )
    return GraphSchemaInferencer(db_flavor=DBType.NEO4J).infer_schema(
        introspection, schema_name=resolved_name
    )

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

The documents of batch with no stored node; see :func:absent_documents.

Source code in graflo/db/neo4j/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: list[Any] | dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """The documents of *batch* with no stored node; see :func:`absent_documents`."""
    return absent_documents(
        self, batch, class_name, match_keys, keep_keys=keep_keys, filters=filters
    )

reset_after_use()

Start a fresh session: closing one finishes any result left pending.

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

def reset_after_use(self) -> None:
    """Start a fresh session: closing one finishes any result left pending."""
    self.conn.close()
    self.conn = self._driver.session(database=self.config.database)

sync_begin(schema, edge_types)

Index every vertex identity, and each edge's owner property.

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

def sync_begin(self, schema: Schema, edge_types: list[SyncEdgeType]) -> None:
    """Index every vertex identity, and each edge's owner property."""
    assert_schema_supported(self.flavor, schema)
    self.bulk_prepare_edges(schema)
    for edge in edge_types:
        if edge.owner_property is not None:
            self._add_index(
                edge.relation_name,
                Index(fields=[edge.owner_property]),
                is_vertex_index=False,
            )
    self._await_indexes()

sync_ensure_vertices(schema, chunk)

MERGE each vertex on its key; only a created one takes the row.

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

def sync_ensure_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
    """``MERGE`` each vertex on its key; only a created one takes the row."""
    self._write(ensure_nodes_query(chunk.class_name, chunk.match_keys), chunk.rows)

sync_write_owners(schema, chunk)

Rewrite the changed owners in one write transaction.

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

def sync_write_owners(self, schema: Schema, chunk: SyncOwnerChunk) -> SyncWritten:
    """Rewrite the changed owners in one write transaction."""
    read = stored_hashes_query(chunk.class_name, chunk.match_keys, chunk.hash_field)
    stored = {
        row["i"]: row["h"]
        for row in _rows(self.execute(read, rows=hash_rows(chunk)))
    }
    changed = changed_owners(chunk, stored)
    statements = owner_statements(chunk, changed)
    if statements:

        def work(tx: Any) -> None:
            for query, params in statements:
                tx.run(cast(LiteralString, query), **params).consume()

        self.conn.execute_write(work)
    return SyncWritten(
        written=len(changed), skipped=len(chunk.owners) - len(changed)
    )

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

Upsert a batch of nodes using Cypher.

Performs an upsert operation on a batch of nodes, using the specified match keys to determine whether to update existing nodes or create new ones.

Parameters:

Name	Type	Description	Default
docs		List of node documents to upsert	required
class_name		Label to upsert into	required
match_keys		Keys to match for upsert operation	required
**kwargs		Additional options: - dry: If True, don't execute the query - union_fields: List properties written as the union of the stored and the given members rather than replaced.	{}

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

def upsert_docs_batch(self, docs, class_name, match_keys, **kwargs):
    """Upsert a batch of nodes using Cypher.

    Performs an upsert operation on a batch of nodes, using the specified
    match keys to determine whether to update existing nodes or create new ones.

    Args:
        docs: List of node documents to upsert
        class_name: Label to upsert into
        match_keys: Keys to match for upsert operation
        **kwargs: Additional options:
            - dry: If True, don't execute the query
            - union_fields: List properties written as the union of the
              stored and the given members rather than replaced.
    """
    dry = kwargs.pop("dry", False)
    union_fields: tuple[str, ...] = tuple(kwargs.pop("union_fields", ()) or ())

    params: dict[str, Any]
    if union_fields:
        q = union_upsert_query(class_name, match_keys, union_fields)
        params = {"data": union_upsert_rows(docs, union_fields)}
    else:
        index_str = ", ".join(
            [f"{cypher_name(k)}: row.{cypher_name(k)}" for k in match_keys]
        )
        q = f"""
        WITH $batch AS batch
        UNWIND batch as row
        MERGE (n:{cypher_name(class_name)} {{ {index_str} }})
        ON MATCH set n += row
        ON CREATE set n += row
    """
        params = {"batch": docs}
    if not dry:
        # A managed write transaction, as for edges: a deadlock with a concurrent
        # write is a transient error the driver retries, and a MERGE is safe to
        # repeat. It has committed, or raised, before the call returns.
        query = cast(LiteralString, q)
        self.conn.execute_write(
            lambda tx, query=query, params=params: tx.run(query, **params).consume()
        )

Functions: