ewox-db-graph

ewoxdbgraph provides the shared Neo4j connection lifecycle and repository infrastructure used by Ewox applications. Domain-specific repositories and Cypher remain in the consuming application.

Installation

poetry add ewoxdbgraph

The package requires Python 3.13 or later and Neo4j Python Driver 6.x.

Engine lifecycle

DBEngine is the package-provided asynchronous engine. An application should create one engine, call setup() during application startup, inject that engine into providers, and call dispose() during shutdown.

from ewoxdbgraph.db_engine import DBEngine
from ewoxdbgraph.db_settings import DBSettings

engine = DBEngine(DBSettings())
await engine.setup()

try:
    # Run the application with this shared engine.
    ...
finally:
    await engine.dispose()

Lifecycle guarantees:

• setup() is idempotent and concurrent setup calls create at most one driver.
• dispose() is idempotent and prevents new sessions from being created.
• A driver created by an unsuccessful or cancelled setup attempt is closed.
• Calling setup() after dispose() creates a fresh driver.

The default settings read BOLT_URL, BOLT_USER, and BOLT_PASSWORD from the environment. Retry count, retry backoff, maximum connection lifetime, and liveness checking can be configured on DBSettings before engine setup.

The package emits lifecycle messages through Python logging under ewoxdbgraph.db_engine. Repository write failures are logged under ewoxdbgraph.repository_base and re-raised unchanged. The package does not install handlers or configure the consuming application's logging policy.

Connection contexts

DBConnection owns one short-lived session and closes it when its context ends. The context syntax must match the session returned by the supplied engine.

The package-provided DBEngine returns neo4j.AsyncSession, so use it with async with:

from ewoxdbgraph.db_connection import DBConnection

async with DBConnection(engine=engine) as session:
    model = await repository.get_model(session, model_id)

DBConnection also supports synchronous consumption for applications that supply a compatible engine whose get_session() returns a synchronous Neo4j session:

with DBConnection(engine=sync_engine) as session:
    model = repository.get_model(session, model_id)

Do not reuse one DBConnection object in nested contexts. Create a new context for each session.

Repository and Cypher contracts

Repositories inherit RepositoryBase and load application-owned Cypher from queries/ and mutations/ directories.

Singular queries

get_item, get_singular_string, get_singular_int, get_singular_float, exists_item, and get_paging_item consume the first column of one returned record. Their Cypher must therefore return one projected value:

MATCH (item:Item {uid: $uid})
RETURN item { .uid, .name }

No row means no item. Query authors should make row cardinality explicit when more than one match is possible.

List queries

get_items extracts a named column from every record. Its default column name is result, and callers can supply another result_name when needed:

MATCH (item:Item)
RETURN item { .uid, .name } AS result

Mutations

save_item, save_entity, execute_write_entity, and each mutation inside save_entities define success as receiving one result row. Application-owned mutation Cypher must return a row only after the intended write path succeeds:

MATCH (item:Item {uid: $uid})
SET item.name = $name
RETURN TRUE AS result

If the query returns no row, the single-mutation helpers return False. save_entities executes all supplied mutations inside one managed transaction and raises if any mutation returns no row.

Sessions and transactions

Reusing one session across a provider workflow reduces session churn and gives the workflow Neo4j's session-level causal ordering. It does not make separate repository calls atomic: every execute_read() or execute_write() call uses its own managed transaction.

Use save_entities or a repository method that performs all required work in one transaction when all-or-nothing database behavior is required.

Development and verification

Install development dependencies and run package-local tests:

poetry install
poetry run pytest -m "not integration"

Real-Neo4j integration tests are opt-in. Provide a disposable test database:

export EWOX_TEST_NEO4J_URL="bolt://localhost:7687"
export EWOX_TEST_NEO4J_USER="neo4j"
export EWOX_TEST_NEO4J_PASSWORD="test-password"
poetry run pytest -m integration

The integration suite does not persist domain data; it executes read and write transactions using constant return values.

Documentation

• Feature overview
• Configuration, engine lifecycle, and connection contexts
• Repository and Cypher contracts
• Testing and continuous integration