Quick Start

Connect to a database

from dpmcore import connect

# SQLite
db = connect("sqlite:///path/to/dpm.db")

# PostgreSQL
db = connect("postgresql://user:pass@host:5432/dpm_db")

# With connection-pool options
db = connect(
    "postgresql://user:pass@host:5432/dpm_db",
    pool_config={"pool_size": 20, "max_overflow": 10},
)

# As a context manager
with connect("sqlite:///dpm.db") as db:
    result = db.services.dpm_xl.validate_syntax("v1 = v2")

Validate a database

Run a shallow shape + seed-data sanity check:

with connect("sqlite:///dpm.db") as db:
    result = db.validate_schema()
    print(result.is_valid)                # True / False
    print(result.missing_tables)          # []
    print(result.empty_required_tables)   # []

The same check is available on the command line and exits non-zero on failure (see CLI Reference):

dpmcore validate --database sqlite:///dpm.db --json

Access services

Read-only DPM dictionary services are available through db.services (full reference: API Reference):

with connect("sqlite:///dpm.db") as db:
    # Syntax validation (no database needed on its own)
    db.services.syntax.validate("{tC_01.00, r0100, c0010}")

    # Semantic validation — release-aware
    db.services.semantic.validate(
        "{tC_01.00, r0100, c0010}",
        release_code="4.2.1",       # or release_id=<int>
    )

    # Data dictionary
    releases = db.services.data_dictionary.get_releases()
    tables   = db.services.data_dictionary.get_tables(release_code="4.2.1")

    # Operation scope calculation
    scope = db.services.scope_calculator.calculate_from_expression(
        expression="{tC_01.00, r0100, c0010} = {tC_01.00, r0200, c0010}",
        release_code="4.2.1",
    )

    # Explorer — reverse lookups
    var = db.services.explorer.get_variable_by_code(
        "mi123", release_code="4.2.1"
    )

    # Framework / module / table tree
    tree = db.services.hierarchy.get_all_frameworks(deep=True)
    details = db.services.hierarchy.get_table_details(
        "C_01.00", date="2024-06-30"
    )
    modelling = db.services.hierarchy.get_table_modelling("C_01.00")

Release-aware methods accept at most one of release_id / release_code / date; passing more than one, or an unknown release_code, raises ValueError. Prefer release_code for human input — IDs became opaque from DPM 4.2.1 onward.

Generate an engine-ready validations script

with connect("sqlite:///dpm.db") as db:
    script = db.services.ast_generator.script(
        expressions=[
            ("{tC_01.00, r0100, c0010} = {tC_01.00, r0200, c0010}", "v0001"),
            ("{tC_01.00, r0200, c0010} > 0", "v0002"),
        ],
        module_code="COREP_Con",
        module_version="2.0.1",
        severity="warning",              # global default
        severities={"v0002": "error"},   # per-validation override
        release="4.2",                   # optional; latest if omitted
    )
    # script["enriched_ast"] is keyed by the resolved module URI;
    # script["failed_operations"] lists expressions skipped due to
    # semantic errors (e.g. grey cells).

The same generation is exposed via dpmcore generate-script and the /api/v1/scripts REST endpoint.

Export table layouts to Excel

from dpmcore.services.layout_exporter.models import ExportConfig

with connect("sqlite:///dpm.db") as db:
    svc = db.services.layout_exporter
    svc.export_module(
        "FINREP9", release_code="4.2", output_path="finrep9.xlsx",
        config=ExportConfig(annotate=True, add_cell_comments=True),
    )

Standalone service usage

Services that don’t require a database can be used directly:

from dpmcore.services import SyntaxService

syntax = SyntaxService()
result = syntax.validate("{tC_01.00, r0100, c0010}")
print(result.is_valid)

MigrationService (which mutates the database) can be used standalone with any SQLAlchemy engine:

from sqlalchemy import create_engine
from dpmcore.loaders.migration import MigrationService

engine = create_engine("sqlite:///new_dpm.db")
service = MigrationService(engine)
result = service.migrate_from_access("/path/to/dpm.accdb")

print(f"Migrated {result.tables_migrated} tables, {result.total_rows} rows")

Direct ORM access

For queries the services don’t cover, use the session directly:

from dpmcore import connect
from dpmcore.orm.infrastructure import Release

with connect("sqlite:///dpm.db") as db:
    releases = db.orm.query(Release).all()