dpmcore.connection

Connection management for standalone library usage.

DpmConnection

class dpmcore.connection.DpmConnection(url, pool_config=None)[source]

Bases: object

A connection to a DPM database.

Holds a SQLAlchemy engine + session and exposes services through the services accessor.

Parameters:

  • url (str) – SQLAlchemy connection URL.

  • pool_config (Optional[Dict[str, Any]]) – Optional engine keyword arguments (pool_size, max_overflow, etc.).

__init__(url, pool_config=None)[source]

Build a connection for url with optional pool_config.

Parameters:
Return type:

None

property orm: Session

Direct access to the SQLAlchemy session.

Use for advanced ORM queries that bypass the service layer.

validate_schema()[source]

Run a shallow shape + data-sanity check on the database.

Reflects the engine, compares against the dpmcore ORM metadata (case-insensitively), and probes a small set of canonical seed tables for non-emptiness. Designed to be cheap (tens of milliseconds on a real DPM database) — not a deep audit.

Return type:

SchemaValidationResult

Returns:

A SchemaValidationResult with the discrepancies. An empty or non-DPM database returns a result with is_valid=False rather than raising.

close()[source]

Close the session and dispose the engine.

Return type:

None

connect

dpmcore.connection.connect(url, pool_config=None)[source]

Open a connection to a DPM database.

Parameters:

  • url (str) – A SQLAlchemy connection URL, e.g. "sqlite:///path/to/dpm.db" or "postgresql://user:pass@host:5432/db".

  • pool_config (Optional[Dict[str, Any]]) – Optional engine keyword arguments.

Return type:

DpmConnection

Returns:

A DpmConnection instance.