# How to Use Neo4j for Custom Queries in Flowsint

> Learn how to execute custom Neo4j queries within Flowsint. Instantiate Neo4jGraphRepository and use the query() method with parameterized Cypher for powerful data retrieval.

- Repository: [reconurge/flowsint](https://github.com/reconurge/flowsint)
- Tags: how-to-guide
- Published: 2026-06-05

---

**To use Neo4j for custom queries in Flowsint, instantiate the `Neo4jGraphRepository` class and call its `query()` method with parameterized Cypher statements, which forwards the request to the underlying Neo4j driver.**

Flowsint persists investigation data in a Neo4j graph database and exposes a high-level Python API for interacting with it. The **Neo4jGraphRepository** class, defined in [`flowsint-core/src/flowsint_core/core/graph/repository.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/repository.py), serves as the primary interface for executing arbitrary Cypher queries that extend beyond the standard node and relationship CRUD operations.

## Neo4jGraphRepository Architecture

The repository abstracts the Neo4j driver behind a singleton connection pattern, ensuring efficient connection reuse across the application while providing a clean Pythonic API for graph operations.

### Core Implementation Details

The **Neo4jGraphRepository** wraps a singleton `Neo4jConnection` instance. When instantiated, it either receives an explicit connection object or obtains the shared singleton via `Neo4jConnection.get_instance()` from the same package. All methods build parameterized Cypher strings and forward them to the driver through `self._connection.query` or transaction-based methods like `execute_write`.

### The query() Method

The generic `query` method (lines 51‑66 in [`repository.py`](https://github.com/reconurge/flowsint/blob/main/repository.py)) serves as the escape hatch for custom Cypher that is not covered by higher-level helpers. It accepts a raw Cypher string and a dictionary of parameters, returning a list of dictionaries representing Neo4j records.

```python
def query(
    self, cypher: str, parameters: Dict[str, Any] = {}
) -> List[Dict[str, Any]]:
    """
    Execute a custom Cypher query.
    """
    if not self._connection:
        return []

    return self._connection.query(cypher, parameters)

```

Source: [[`flowsint-core/src/flowsint_core/core/graph/repository.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/repository.py)](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/repository.py#L51-L66)

Because the repository is utilized throughout the core service layer in [`flowsint-core/src/flowsint_core/core/graph/service.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/service.py), you can invoke `query` directly on a repository instance or obtain it via the higher-level **Neo4jGraphService**, which mirrors the same method signature.

## Executing Custom Cypher Queries

When using Neo4j for custom queries in Flowsint, follow this workflow: obtain a repository instance, write parameterized Cypher using `$param` syntax, pass a dictionary of parameters, and process the returned list of dictionaries. Each dictionary corresponds to a Neo4j record where keys match the `RETURN` clause aliases.

The repository automatically respects **sketch isolation** (`sketch_id`) and **soft-delete semantics** (`deleted_at IS NULL`). Custom queries should explicitly include these filters when operating on investigation data to maintain data integrity.

## Practical Implementation Examples

### Querying Nodes by Label

The following example demonstrates a simple node lookup using custom Cypher against the repository:

```python
from flowsint_core.core.graph import Neo4jGraphRepository

repo = Neo4jGraphRepository()          # uses the singleton connection

cypher = """
    MATCH (n:Domain {nodeLabel: $label, sketch_id: $sketch_id})
    RETURN elementId(n) AS id, properties(n) AS data
"""
params = {"label": "example.com", "sketch_id": "sketch-123"}

results = repo.query(cypher, params)

# results → [{'id': '12345', 'data': {'nodeLabel': 'example.com', ...}}]

```

### Batch Updating Properties

For efficient batch operations, use the `UNWIND` clause to process multiple updates in a single transaction:

```python
cypher = """
UNWIND $updates AS upd
MATCH (n) WHERE elementId(n) = upd.id AND n.sketch_id = $sketch_id
SET n += upd.props
RETURN count(n) AS updated
"""
params = {
    "sketch_id": "sketch-123",
    "updates": [
        {"id": "12345", "props": {"confidence": 0.9}},
        {"id": "67890", "props": {"confidence": 0.85}},
    ],
}
result = repo.query(cypher, params)
print(result[0]["updated"])   # → 2

```

### Executing Soft Deletes

Flowsint uses soft deletes rather than hard removals. Custom queries should respect this pattern by setting the `deleted_at` timestamp:

```python
cypher = """
OPTIONAL MATCH (n {sketch_id: $sketch_id})
WHERE n.deleted_at IS NULL
SET n.deleted_at = datetime()
RETURN count(n) AS soft_deleted
"""
result = repo.query(cypher, {"sketch_id": "sketch-123"})
print(f"Soft‑deleted {result[0]['soft_deleted']} nodes")

```

### FastAPI Endpoint Integration

In a web service context, use FastAPI dependency injection to obtain the repository:

```python
from fastapi import APIRouter, Depends
from flowsint_core.core.graph.service import Neo4jGraphRepository

router = APIRouter()

@router.get("/custom")
def run_custom_query(
    repo: Neo4jGraphRepository = Depends(),
    label: str = "example.com",
    sketch_id: str = "sketch-123",
):
    cypher = """
        MATCH (n {nodeLabel: $label, sketch_id: $sketch_id})
        RETURN elementId(n) AS id, properties(n) AS data
    """
    return repo.query(cypher, {"label": label, "sketch_id": sketch_id})

```

## Security and Data Integrity Considerations

Always use **parameterized queries** with the `$parameter` syntax rather than Python f-strings or concatenation to prevent Cypher injection attacks. The `query` method passes parameters directly to the Neo4j driver without modification, ensuring safe execution.

When writing custom queries, include `sketch_id` filters to maintain multi-tenancy boundaries between investigations. Additionally, check for `deleted_at IS NULL` unless your specific use case requires accessing archived data.

## Summary

- **Neo4jGraphRepository** in [`flowsint-core/src/flowsint_core/core/graph/repository.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/repository.py) provides the `query()` method for executing arbitrary Cypher.
- The method signature accepts a Cypher string and parameter dictionary, returning a list of record dictionaries.
- The repository manages a singleton connection via `Neo4jConnection` to optimize driver performance.
- Custom queries must respect `sketch_id` isolation and `deleted_at` soft-delete semantics for data consistency.
- Integration with FastAPI uses standard dependency injection patterns against the repository or service layer.

## Frequently Asked Questions

### How do I access the raw Neo4j driver for advanced configurations?

You generally should not access the raw driver directly. Instead, instantiate **Neo4jGraphRepository**, which encapsulates the connection logic in [`flowsint-core/src/flowsint_core/core/graph/connection.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/connection.py). If you need specific driver features, extend the repository class rather than bypassing it, as the singleton pattern ensures proper connection pooling and transaction management across the application.

### What is the difference between Neo4jGraphRepository and Neo4jGraphService?

**Neo4jGraphRepository** provides low-level database access and the generic `query()` method for custom Cypher. **Neo4jGraphService**, defined in [`flowsint-core/src/flowsint_core/core/graph/service.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/graph/service.py), composes the repository and adds domain-specific business logic, validation, and orchestration. For custom queries, you can use either class, as the service exposes the same `query` signature while adding higher-level abstractions for standard operations.

### Do I need to manually close connections when using the query method?

No. The repository utilizes a singleton `Neo4jConnection` that manages the Neo4j driver lifecycle automatically. When you call `repo.query()`, the underlying connection remains open for subsequent operations, and the driver handles connection pooling internally. This design prevents connection leaks and reduces latency for repeated queries within the same process.

### How should I index properties for performance in custom queries?

Flowsint creates indexes on commonly queried properties such as `sketch_id`, as shown in `neo4j-migrations/001_indexes.cypher`. When writing custom queries that filter on specific properties, ensure those properties are indexed in your Neo4j schema to avoid expensive full-graph scans. The repository does not automatically create indexes for custom query parameters, so database optimization remains the responsibility of the developer.