dpmcore.services.expression_metadata

Resolve the DPM entities (tables, headers, frameworks) that a DPM-XL expression references. Where ScopeCalculatorService answers “which module versions does this expression touch?”, this service answers “given that expression, which concrete tables / headers / frameworks do I need to persist alongside it?”.

Callers get plain list[dict] back — no ORM instances leak — so the results are safe to hand to a downstream ORM or serializer without holding onto the SQLAlchemy session.

Usage

from dpmcore import connect

with connect("sqlite:///dpm.db") as db:
    svc = db.services.expression_metadata

    tables = svc.get_referenced_tables(
        expression="{tF_01.01, r0010, c0010} = 100",
        release_id=42,
    )
    headers = svc.get_referenced_headers(
        expression="{tF_01.01, r0010, c0010} = 100",
        release_id=42,
    )
    frameworks = svc.get_referenced_frameworks(
        expression="{tF_01.01, r0010, c0010} = 100",
        release_id=42,
    )

The header_type field on each header entry reflects the header’s use in the expression (r* → "Row", c* → "Column", s* → "Sheet"), not the catalog Header.direction — two tables that share a code but are transposed relative to each other both emit rows whose header_type matches the syntax the expression used.

Any parser or semantic failure degrades to an empty list rather than raising.

ExpressionMetadataService

class dpmcore.services.expression_metadata.ExpressionMetadataService(session)

Bases: object

Resolve DPM entities referenced by a DPM-XL expression.

The service parses the expression once (via SyntaxService + OperandsChecking) and then queries the ORM to hydrate the referenced tables, headers, and frameworks.

Parameters:

session (Session) – An open SQLAlchemy session.

__init__(session)

Build the service bound to session.

Parameters:

session (Session)

Return type:

None

get_referenced_tables(expression, release_id=None, release_code=None)

Return the tables referenced by expression.

Each entry is a plain dict with the keys table_vid, code, name, description, module_vid, module_code, module_name, module_version.

Result is sorted by (code, table_vid, module_vid) so callers can diff-persist deterministically.

On a syntax/semantic error (or when the expression references nothing that resolves), returns [].

Return type:

List[Dict[str, Any]]

Parameters:

• expression (str)

• release_id (int | None)

• release_code (str | None)

get_referenced_headers(expression, release_id=None, table_vid=None, release_code=None)

Return the headers referenced by expression.

Each entry is a plain dict with header_vid, code, label, header_type (one of "Row", "Column", "Sheet"), table_vid, table_code, table_name.

header_type reflects the header’s use in the expression (r*/c*/s*), not the catalog Header.direction: two tables that share a header code but are transposed relative to each other both emit rows whose header_type matches the syntax the expression used.

table_vid, when passed, narrows the result to headers of that specific table version.

Result is sorted deterministically by (table_vid, header_type, code, header_vid).

On a syntax/semantic error, returns [].

Return type:

List[Dict[str, Any]]

Parameters:

• expression (str)

• release_id (int | None)

• table_vid (int | None)

• release_code (str | None)

get_referenced_frameworks(expression, release_id=None, release_code=None)

Return the frameworks touched by expression.

Each entry is a plain dict with framework_id, code, name, description. Result is deduped and sorted by (code, framework_id).

On a syntax/semantic error, returns [].

Return type:

List[Dict[str, Any]]

Parameters:

• expression (str)

• release_id (int | None)

• release_code (str | None)