"""Operation scope calculation service."""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from types import SimpleNamespace
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Mapping,
Optional,
Set,
Tuple,
Union,
)
from dpmcore.dpm_xl.ast.operands import OperandsChecking
from dpmcore.dpm_xl.utils.filters import resolve_release_id
from dpmcore.dpm_xl.utils.scopes_calculator import (
OperationScopeService,
)
from dpmcore.errors import SemanticError
from dpmcore.orm.glossary import Property
from dpmcore.orm.infrastructure import DataType, Release
from dpmcore.orm.operations import OperationScope, OperationScopeComposition
from dpmcore.orm.packaging import (
ModuleVersion,
ModuleVersionComposition,
)
from dpmcore.orm.query_utils import chunked_in
from dpmcore.orm.rendering import (
TableVersion,
TableVersionCell,
)
from dpmcore.orm.variables import Variable, VariableVersion
from dpmcore.services._open_keys import (
get_open_keys_for_tables as _get_open_keys_for_tables,
)
from dpmcore.services._precondition_codes import required_precondition_codes
from dpmcore.services.syntax import SyntaxService
if TYPE_CHECKING:
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
class UnsupportedDbScope(Exception):
"""Raised when no persisted, non-phantom scope exists for an operation.
Always a request to fall back to
:meth:`ScopeCalculatorService.calculate_from_expression` for this
one operation — never a signal that the DB data itself is broken.
"""
[docs]
@dataclass
class ScopeResult:
"""Outcome of a scope calculation.
``warning`` is only ever set when a ``precondition_expression`` changed the
computed scope. ``error_source`` names the half a failure belongs to and is
set whenever ``has_error`` is true. Neither changes ``error_message`` for a
call that passes no precondition expression.
"""
scopes: list[Any] = field(default_factory=list)
total_scopes: int = 0
is_cross_module: bool = False
module_versions: List[int] = field(default_factory=list)
has_error: bool = False
error_message: Optional[str] = None
# Set when the precondition expression changes the computed scope.
warning: Optional[str] = None
# Which half a failure belongs to: "expression" or "precondition".
error_source: Optional[str] = None
[docs]
class ScopeCalculatorService:
"""Calculate operation scopes for DPM-XL expressions.
Determines which module versions are involved in an operation
based on table references and precondition items.
Args:
session: An open SQLAlchemy session.
"""
[docs]
def __init__(self, session: "Session") -> None:
"""Build the service bound to ``session``."""
self.session = session
self._syntax = SyntaxService()
# ``(ModuleVersion, URI)`` per module VID, for the home module of
# a script. ``detect_cross_module_dependencies`` runs once per
# operation with a fixed ``primary_module_vid``, so resolving it
# inside repeats the same row fetch and URI resolution for every
# operation that shifts a home table. Both are keyed by VID
# alone and neither varies with the release being generated.
self._home_module_refs: Dict[int, Tuple[Any, Optional[str]]] = {}
# {release_id: {ghost_module_vid: fallback_module_vid}} (#182),
# release-wide like ast_generator.py's own
# ``_ghost_fallback_cache`` — :meth:`build_scope_result_from_db`
# computes it once per *operation*, so memoising matters here
# too.
self._ghost_fallback_map_cache: Dict[int, Dict[int, int]] = {}
# {module_vid: [ghost_vid, ...]} (#182) — the ghost siblings a
# module version's own window bridges, memoised per module_vid
# since :meth:`build_scope_result_from_db` recomputes it once
# per operation for the same, fixed ``primary_module_vid``.
self._ghost_chain_vids_cache: Dict[int, List[int]] = {}
def _check_release_exists(self, release_id: Optional[int]) -> None:
"""Raise SemanticError if *release_id* does not exist."""
if release_id is None:
return
exists = (
self.session.query(Release.release_id)
.filter(Release.release_id == release_id)
.first()
)
if exists is None:
raise SemanticError("1-21", release_id=release_id)
@staticmethod
def _compute_cross_module(scopes: list[Any]) -> bool:
"""Return True if any scope spans more than one module."""
return any(
len(
{
c.module_vid
for c in getattr(s, "operation_scope_compositions", [])
}
)
> 1
for s in scopes
)
@staticmethod
def _scope_signature(scopes: list[Any]) -> frozenset[frozenset[int]]:
"""Canonical identity of a scope set: the set of module-VID sets.
Comparing ``module_versions`` alone would miss a re-partitioning —
two intra-module scopes ``[[462], [513]]`` collapsing into one
cross-module scope ``[[462, 513]]`` lists the same module versions but
means something materially different. Comparing ``total_scopes`` alone
would miss a coincidental equal count. The set of module sets catches
module additions and removals, intra/cross re-grouping, splits, and the
empty case.
"""
return frozenset(
frozenset(
c.module_vid
for c in getattr(s, "operation_scope_compositions", [])
)
for s in scopes
)
@staticmethod
def _module_vids(scopes: list[Any]) -> List[int]:
"""Module VIDs across *scopes*, deduped in first-seen order."""
mvids: List[int] = []
for scope in scopes:
for comp in getattr(scope, "operation_scope_compositions", []):
if comp.module_vid not in mvids:
mvids.append(comp.module_vid)
return mvids
def _result_from_scopes(self, scopes: list[Any]) -> ScopeResult:
"""Project a computed scope list into a :class:`ScopeResult`."""
return ScopeResult(
scopes=scopes,
total_scopes=len(scopes),
is_cross_module=self._compute_cross_module(scopes),
module_versions=self._module_vids(scopes),
)
def _operand_tables(
self, expression: str, release_id: Optional[int]
) -> tuple[Any, List[str]]:
"""Parse *expression*, returning its AST and its table codes."""
ast = self._syntax.parse(expression)
oc = OperandsChecking(
session=self.session,
expression=expression,
ast=ast,
release_id=release_id,
)
return ast, (list(oc.tables.keys()) if oc.tables else [])
def _check_tables_hosted(
self, table_codes: List[str], release_id: Optional[int]
) -> None:
"""Raise ``1-13`` naming any table code no module version hosts.
``OperationScopeService.extract_module_info`` only raises ``1-13`` when
*nothing* resolves, so an unhostable gate table is otherwise absorbed
by the main expression's own rows. It still inflates the operand count,
which usually empties the scope — but when the main expression resolves
to a single module-info row the resolver short-circuits before counting
operands at all, and the call would return a scope that cannot in fact
evaluate the pair. Checking the gate's codes on their own keeps that
case an error attributed to the gate rather than a silent wrong answer.
"""
if not table_codes:
return
from dpmcore.dpm_xl.model_queries import ModuleVersionQuery
df = ModuleVersionQuery.get_from_table_codes(
session=self.session,
table_codes=table_codes,
release_id=release_id,
)
hosted = (
set(df["TableCode"].dropna().unique()) if not df.empty else set()
)
missing = [code for code in table_codes if code not in hosted]
if missing:
raise SemanticError("1-13", table_version_ids=missing)
def _run_scope(
self,
table_codes: List[str],
precondition_items: List[str],
release_id: Optional[int],
) -> list[Any]:
"""Resolve one scope set. A fresh service per call — it accumulates."""
scopes, _ = OperationScopeService(
session=self.session
).calculate_operation_scope(
tables_vids=[],
precondition_items=precondition_items,
release_id=release_id,
table_codes=table_codes,
)
return scopes
@staticmethod
def _scope_change_warning(
baseline: list[Any],
combined: list[Any],
gate_tables: List[str],
gate_items: List[str],
) -> str:
"""Describe how the precondition moved the scope."""
before = sorted(ScopeCalculatorService._module_vids(baseline))
after = sorted(ScopeCalculatorService._module_vids(combined))
contributed = ", ".join(
part
for part in (
f"tables {', '.join(gate_tables)}" if gate_tables else "",
(
f"filing indicators {', '.join(gate_items)}"
if gate_items
else ""
),
)
if part
)
message = (
"Precondition expression changes the scope of the expression: "
f"module versions {before} -> {after} "
f"(scopes {len(baseline)} -> {len(combined)})."
)
if not combined:
message += (
" The pair is not evaluable in any module version:"
" no module reports every operand both halves need."
)
if contributed:
message += f" Precondition operands: {contributed}."
return message
[docs]
def calculate_from_expression(
self,
expression: str,
release_id: Optional[int] = None,
precondition_items: Optional[List[str]] = None,
release_code: Optional[str] = None,
*,
precondition_expression: Optional[str] = None,
) -> ScopeResult:
"""Calculate scopes for *expression*, optionally gated.
Parses the expression, runs OperandsChecking to extract table
codes, then delegates to :class:`OperationScopeService`.
``precondition_items`` is the list of precondition variable
codes that gate the validation; pass ``None`` or ``[]`` if
the validation has no preconditions.
When ``precondition_expression`` is supplied, the gate's own operands
join the resolution by their matching channels — its table codes are
unioned into ``table_codes`` and its *mandatory* precondition variable
codes into ``precondition_items`` (unioned with any the caller passed).
The two are not interchangeable: only ``table_codes`` resolves through
``get_from_table_codes``, and only ``precondition_items`` gets the
filing-indicator filter and the ``1-14`` check. Because a module hosts
an intra-module scope only when it supplies *every* operand, a gate
reaching outside the expression's own modules widens the scope to
cross-module, or empties it — which is the honest verdict, since the
pair is only evaluable where both halves resolve. That change is
reported in ``warning``, and a gate-attributable failure sets
``error_source`` to ``"precondition"`` with a prefixed message.
The two channels treat a disjunctive gate differently, deliberately.
Variable codes are intersected across ``or`` branches, so an optional
filing indicator does not constrain scope. Table codes come from the
gate's ``OperandsChecking`` pass, which does not model boolean
structure, so every table the gate mentions is required even when only
one branch needs it. That errs toward reporting a wider scope rather
than missing a table the pair genuinely needs, and costs nothing on the
real dictionary, where no persisted precondition references a table.
Passing no ``precondition_expression`` costs nothing: no second scope
resolution, no warning, and byte-identical error messages.
Args:
expression: The DPM-XL expression to scope.
release_id: Optional release ID filter.
precondition_items: Filing-indicator codes gating the validation.
release_code: Optional release code (mutually exclusive
with ``release_id``).
precondition_expression: Optional DPM-XL gate expression.
Keyword-only, matching ``SemanticService.validate``, so the
pre-existing positional arguments keep their meaning.
"""
base_items = list(precondition_items or [])
try:
release_id = resolve_release_id(
self.session,
release_id=release_id,
release_code=release_code,
)
self._check_release_exists(release_id)
_, main_tables = self._operand_tables(expression, release_id)
except Exception as exc:
return self._failed(exc, None)
if precondition_expression is None:
return self._scope_or_error(main_tables, base_items, release_id)
try:
gate_ast, gate_tables = self._operand_tables(
precondition_expression, release_id
)
self._check_tables_hosted(gate_tables, release_id)
except Exception as exc:
return self._failed(exc, "precondition")
gate_items = required_precondition_codes(gate_ast)
table_codes = main_tables + [
t for t in gate_tables if t not in main_tables
]
items = base_items + [i for i in gate_items if i not in base_items]
# A gate that contributed no new operand cannot move the scope, so the
# baseline is provably identical and never computed.
if set(table_codes) == set(main_tables) and set(items) == set(
base_items
):
return self._scope_or_error(main_tables, base_items, release_id)
try:
combined = self._run_scope(table_codes, items, release_id)
except Exception as exc:
# The gate is only to blame if the expression scopes cleanly
# without it — settled by retrying, not guessed at.
baseline_ok = self._resolves(main_tables, base_items, release_id)
return self._failed(exc, "precondition" if baseline_ok else None)
result = self._result_from_scopes(combined)
try:
baseline = self._run_scope(main_tables, base_items, release_id)
except Exception:
# The expression alone does not resolve but the pair does; treat
# the baseline as empty so the change is still reported.
baseline = []
if self._scope_signature(baseline) != self._scope_signature(combined):
result.warning = self._scope_change_warning(
baseline, combined, gate_tables, gate_items
)
return result
@staticmethod
def _failed(exc: Exception, source: Optional[str]) -> ScopeResult:
"""Build a failing result, attributed to *source* when known.
``source`` is ``None`` for a failure that is the expression's own (or
that no precondition was involved in), which keeps ``error_message``
byte-identical to what callers saw before this argument existed.
"""
message = str(exc)
if source == "precondition":
message = f"Precondition: {message}"
return ScopeResult(
has_error=True,
error_message=message,
error_source="expression" if source is None else source,
)
def _scope_or_error(
self,
table_codes: List[str],
items: List[str],
release_id: Optional[int],
) -> ScopeResult:
"""Resolve one scope set into a result, or an unattributed failure."""
try:
return self._result_from_scopes(
self._run_scope(table_codes, items, release_id)
)
except Exception as exc:
return self._failed(exc, None)
def _resolves(
self,
table_codes: List[str],
items: List[str],
release_id: Optional[int],
) -> bool:
"""True when these operands resolve without raising."""
try:
self._run_scope(table_codes, items, release_id)
except Exception:
return False
return True
[docs]
def calculate_from_tables(
self,
table_vids: List[int],
precondition_items: Optional[List[str]] = None,
release_id: Optional[int] = None,
table_codes: Optional[List[str]] = None,
release_code: Optional[str] = None,
) -> ScopeResult:
"""Calculate scopes directly from table version IDs."""
try:
release_id = resolve_release_id(
self.session,
release_id=release_id,
release_code=release_code,
)
self._check_release_exists(release_id)
scope_svc = OperationScopeService(session=self.session)
scopes, _ = scope_svc.calculate_operation_scope(
tables_vids=table_vids,
precondition_items=precondition_items or [],
release_id=release_id,
table_codes=table_codes,
)
mvids: List[int] = []
for scope in scopes:
for comp in getattr(scope, "operation_scope_compositions", []):
vid = comp.module_vid
if vid not in mvids:
mvids.append(vid)
return ScopeResult(
scopes=scopes,
total_scopes=len(scopes),
is_cross_module=self._compute_cross_module(scopes),
module_versions=mvids,
)
except Exception as exc:
return ScopeResult(
has_error=True,
error_message=str(exc),
)
[docs]
def build_scope_result_from_db(
self,
operation_vid: int,
primary_module_vid: int,
release_id: Optional[int] = None,
) -> ScopeResult:
"""Rebuild a ``ScopeResult`` for *operation_vid* from the DB.
Reads every active ``OperationScope`` row for the operation —
each already carrying its ``OperationScopeComposition`` module
VIDs via the ORM relationship — instead of recomputing scope
by re-parsing the expression text and re-validating every
operand against live table data via ``OperandsChecking``. That
re-validation is what a genuinely persisted validation's cell
references can legitimately fail on (a wildcard/range axis
resolved against a table that has since grown grey cells the
validation's author never touched, for instance) — none of
which mdpm's own export ever sees, since it reads the same
persisted scope this reads and never re-validates it against
live table data either.
Excludes any scope whose only *other* composed module (besides
*primary_module_vid*) is a #182 ghost with no fallback at
*release_id* (see :meth:`_phantom_paired_scope_ids`), matching
:meth:`~dpmcore.services.ast_generator.ASTGeneratorService.\
_discover_module_validations`'s own exclusion. Applied whenever a
*release_id* is given (whether resolved from an explicit
``--release`` or auto-picked for ``--module-version``/
``--all-versions``) — per issue #182: "this fallback must apply
consistently on every path dpmcore uses to resolve which module
version applies". mdpm's own reference has no equivalent to
#182 in any mode, so it is not a valid comparison here.
When a composition names a ghost that has a fallback, it is not
excluded, but its ``module_vid`` is substituted for the
fallback's (see :meth:`_substitute_ghost_compositions`) — the
persisted composition names the ghost directly, but the ghost
itself has no table/URI structure for downstream dependency
resolution to use, only the fallback does.
Raises :class:`UnsupportedDbScope` when nothing usable
survives — callers must catch this and fall back to
:meth:`calculate_from_expression` instead.
"""
scopes = (
self.session.query(OperationScope)
.filter(
OperationScope.operation_vid == operation_vid,
OperationScope.is_active.in_([-1, 1, True]),
)
.all()
)
if not scopes:
raise UnsupportedDbScope(
f"No active OperationScope rows for "
f"operation_vid={operation_vid}"
)
# The release-wide map only recognises a ghost as having a
# fallback when the ghost's *own* window covers this exact
# release — a chain bridging more than one ghost (e.g.
# COREP_FRTB's 3.2.0 then 3.3.0) leaves an earlier one
# unrecognised there even though it genuinely stands in for
# primary_module_vid too. Folding primary_module_vid's own
# ghost chain in directly closes that gap without making the
# release-wide map itself chain-aware.
ghost_fallback_map: Dict[int, int] = {
**(
self._ghost_fallback_map(release_id)
if release_id is not None
else {}
),
**dict.fromkeys(
self._ghost_chain_vids(primary_module_vid), primary_module_vid
),
}
scope_ids = {s.operation_scope_id for s in scopes}
phantom_ids = self._phantom_paired_scope_ids(
self.session, scope_ids, primary_module_vid, ghost_fallback_map
)
filtered = [
s for s in scopes if s.operation_scope_id not in phantom_ids
]
if not filtered:
raise UnsupportedDbScope(
f"Every OperationScope for operation_vid={operation_vid} "
"is phantom-paired"
)
substituted = (
[
self._substitute_ghost_compositions(s, ghost_fallback_map)
for s in filtered
]
if ghost_fallback_map
else filtered
)
return ScopeResult(
scopes=substituted,
total_scopes=len(substituted),
is_cross_module=self._compute_cross_module(substituted),
module_versions=self._module_vids(substituted),
)
def _ghost_fallback_map(self, release_id: int) -> Dict[int, int]:
"""``{ghost_module_vid: fallback_module_vid}`` for *release_id*.
Inverts :func:`ModuleVersionQuery.ghost_fallbacks` (#182): a
fallback can stand in for several ghosts, but a ghost has at
most one fallback per release, so the inversion is total.
Memoised per release — :meth:`build_scope_result_from_db` and
:meth:`_phantom_paired_scope_ids` both compute it once per
*operation*, and it's release-wide (same reasoning as
ast_generator.py's own ``_release_ghost_fallbacks`` cache).
"""
cached = self._ghost_fallback_map_cache.get(release_id)
if cached is None:
from dpmcore.dpm_xl.model_queries import ModuleVersionQuery
fallbacks = ModuleVersionQuery.ghost_fallbacks(
self.session, release_id
)
cached = {
ghost_vid: fallback_vid
for fallback_vid, ghost_vids in fallbacks.items()
for ghost_vid in ghost_vids
}
self._ghost_fallback_map_cache[release_id] = cached
return cached
def _ghost_chain_vids(self, module_vid: int) -> List[int]:
"""Every #182 ghost sibling in *module_vid*'s own contiguous chain.
Mirrors :meth:`~dpmcore.services.ast_generator.\
ASTGeneratorService._ghosts_represented_by` — the same contiguous run
of ghost siblings :meth:`~dpmcore.services.ast_generator.\
ASTGeneratorService._effective_end_release_id` extends a module
version's own window through, not just whichever one ghost a
release-wide lookup happens to resolve. Duplicated here rather
than shared: the walk only needs ``session`` and a
``ModuleVersion`` row, and this service has no dependency on
``ASTGeneratorService`` otherwise. Memoised per *module_vid*.
"""
cached = self._ghost_chain_vids_cache.get(module_vid)
if cached is not None:
return cached
from dpmcore.orm.release_sort_order import resolve_sort_order
mv = (
self.session.query(ModuleVersion)
.filter(ModuleVersion.module_vid == module_vid)
.one_or_none()
)
if mv is None or mv.end_release_id is None or mv.module_id is None:
self._ghost_chain_vids_cache[module_vid] = []
return []
session = self.session
end_sort = resolve_sort_order(
session, mv.end_release_id, role="module version end release"
)
siblings = (
session.query(ModuleVersion)
.filter(ModuleVersion.module_id == mv.module_id)
.filter(ModuleVersion.module_vid != mv.module_vid)
.all()
)
candidates = self._ghost_chain_candidates(siblings, end_sort)
ghost_vids: List[int] = []
boundary = end_sort
for start_sort, is_ghost, sibling in candidates:
if start_sort > boundary or not is_ghost:
break
ghost_vids.append(sibling.module_vid)
if sibling.end_release_id is None:
break
sibling_end_sort = resolve_sort_order(
session,
sibling.end_release_id,
role="sibling module version end release",
)
if sibling_end_sort > boundary:
boundary = sibling_end_sort
self._ghost_chain_vids_cache[module_vid] = ghost_vids
return ghost_vids
def _ghost_chain_candidates(
self, siblings: List[Any], end_sort: int
) -> List[Tuple[int, bool, Any]]:
"""Sibling ``ModuleVersion`` rows past *end_sort*, sorted by start.
Split out of :meth:`_ghost_chain_vids` to keep its own walk
within the complexity limit.
"""
from dpmcore.orm.release_sort_order import resolve_sort_order
session = self.session
def sort_or_none(
release_id: Optional[int], role: str
) -> Optional[int]:
if release_id is None:
return None
return resolve_sort_order(session, release_id, role=role)
candidates: List[Tuple[int, bool, Any]] = []
for sibling in siblings:
sibling_end_sort = sort_or_none(
sibling.end_release_id, "sibling module version end release"
)
if sibling_end_sort is not None and sibling_end_sort <= end_sort:
continue
sibling_start_sort = sort_or_none(
sibling.start_release_id,
"sibling module version start release",
)
effective_start = (
end_sort
if sibling_start_sort is None
else max(sibling_start_sort, end_sort)
)
is_ghost = (
sibling.from_reference_date is not None
and sibling.to_reference_date is not None
and sibling.from_reference_date == sibling.to_reference_date
)
candidates.append((effective_start, is_ghost, sibling))
candidates.sort(key=lambda item: item[0])
return candidates
@staticmethod
def _substitute_ghost_compositions(
scope: Any, ghost_fallback_map: Dict[int, int]
) -> Any:
"""Return *scope* with any #182-ghost composition substituted.
Substitutes a ghost composition's ``module_vid`` for its
fallback's. A composition naming a ghost module directly is a real,
persisted fact — #182's own ``OperationScopeComposition`` rows
are written against the ghost, never against the fallback that
represents it later — but the ghost has no table/URI structure
of its own for downstream dependency resolution
(``_compute_cross_module``, ``_module_vids``,
``filter_valid_dependency_modules``,
``detect_cross_module_dependencies``) to use. Every one of
those reads ``operation_scope_compositions`` off a scope via
``getattr`` with a default — already duck-typed — so a
lightweight, non-persisted substitute is enough; nothing here
touches the real ORM rows or the session.
"""
compositions = getattr(scope, "operation_scope_compositions", [])
if not any(c.module_vid in ghost_fallback_map for c in compositions):
return scope
return SimpleNamespace(
operation_scope_id=scope.operation_scope_id,
operation_scope_compositions=[
SimpleNamespace(
operation_scope_id=c.operation_scope_id,
module_vid=ghost_fallback_map.get(
c.module_vid, c.module_vid
),
row_guid=getattr(c, "row_guid", None),
)
for c in compositions
],
)
@staticmethod
def _phantom_paired_scope_ids(
session: "Session",
scope_ids: Set[int],
module_vid: int,
ghost_fallback_map: Dict[int, int],
) -> Set[int]:
"""Return scope ids whose only other composed module is a dead end.
These are the ``OperationScopeID``s whose only *other* composed
module versions (besides *module_vid*) are all phantom (#182
ghosts — mirrors mdpm's ``is_phantom_module_op``) **and** have
no fallback standing in for them in *ghost_fallback_map*. A
phantom *with* a fallback there is not void: the fallback
substitutes for it (see :meth:`_substitute_ghost_compositions`),
so a scope composed with such a ghost is kept.
*ghost_fallback_map* is empty whenever no release could be
resolved at all, which reduces this to the original rule (every
phantom is a dead end). A scope shared with no other module at
all is never phantom here — this only
voids a cross-framework pairing whose counterpart neither
really existed as its own module version, nor has anything
current standing in for it.
Shared by :meth:`build_scope_result_from_db` and
:meth:`~dpmcore.services.ast_generator.ASTGeneratorService.\
_discover_module_validations`.
"""
rows = (
session.query(
OperationScopeComposition.operation_scope_id,
OperationScopeComposition.module_vid,
ModuleVersion.from_reference_date,
ModuleVersion.to_reference_date,
)
.join(
ModuleVersion,
ModuleVersion.module_vid
== OperationScopeComposition.module_vid,
)
.filter(
OperationScopeComposition.operation_scope_id.in_(scope_ids)
)
.filter(OperationScopeComposition.module_vid != module_vid)
.all()
)
others_by_scope: Dict[int, List[bool]] = {}
for scope_id, other_module_vid, from_date, to_date in rows:
is_phantom = (
from_date is not None
and to_date is not None
and from_date == to_date
)
is_dead_end = (
is_phantom and other_module_vid not in ghost_fallback_map
)
others_by_scope.setdefault(scope_id, []).append(is_dead_end)
return {
scope_id
for scope_id, flags in others_by_scope.items()
if flags and all(flags)
}
# ------------------------------------------------------------------ #
# Cross-module dependency detection (Fix 2)
# ------------------------------------------------------------------ #
[docs]
def filter_valid_dependency_modules(
self,
scope_result: ScopeResult,
primary_module_vid: int,
) -> Set[int]:
"""Return module VIDs that co-occur with *primary_module_vid*.
Only modules that actually appear alongside the primary module
in a multi-module scope are valid cross-module partners.
This filters out sibling modules that share tables but are
not actual cross-module dependencies.
"""
valid: Set[int] = set()
for scope in scope_result.scopes or []:
scope_vids = {
c.module_vid
for c in getattr(scope, "operation_scope_compositions", [])
}
if primary_module_vid in scope_vids and len(scope_vids) > 1:
valid.update(scope_vids - {primary_module_vid})
return valid
[docs]
def detect_cross_module_dependencies(
self,
scope_result: ScopeResult,
primary_module_vid: int,
operation_code: Optional[str] = None,
release_id: Optional[int] = None,
time_shifts: Optional[Mapping[str, Union[str, List[str]]]] = None,
compute_alternative_deps: bool = True,
release_code: Optional[str] = None,
referenced_variables: Optional[Dict[str, str]] = None,
referenced_tables: Optional[Set[str]] = None,
home_module_tables: Optional[Set[str]] = None,
) -> Dict[str, Any]:
"""Build dependency information for a scope result.
Args:
scope_result: The computed scope result.
primary_module_vid: VID of the primary module.
operation_code: Current operation code (if any).
release_id: Optional release filter.
time_shifts: Optional mapping of table codes to the
ref-period strings that table is read at (e.g.
``{"C_01.00": ["T", "T-1Q"]}``). A table read at more
than one period needs one declared instance per period
(#326). A bare string is accepted as a single period, so
the pre-#326 ``{"C_01.00": "T-1Q"}`` shape still works.
Tables not present default to ``["T"]``.
compute_alternative_deps: When True (default) the returned
``alternative_dependencies`` is populated from this
single ``scope_result``. Aggregating callers that
compute alternatives across many scope results should
pass ``False`` to avoid the per-call work.
release_code: Optional release code; resolved to
``release_id`` via :class:`Release.code`. Mutually
exclusive with ``release_id``.
referenced_variables: Optional ``{datapoint: type_code}`` of
every operand datapoint the operation references, across
all modules it spans — home module included. Declared in
each dependency module's ``variables`` map (#251).
referenced_tables: Optional table codes the operation
references. Together with ``referenced_variables`` this
narrows each dependency module's declaration to the
subset the operation uses (#250); omit both to declare
the dependency modules whole.
home_module_tables: Optional pre-computed set of table codes
owned by the primary (home) module — the same set this
method would derive from :meth:`_get_module_tables`.
Callers that iterate this method with a fixed
``primary_module_vid`` (per-op dependency detection
over a script's operations) should pass a single
pre-computed set to avoid re-running the per-table
variable/open-key fetch on every iteration.
Returns a dict with:
- ``intra_instance_validations``
- ``cross_instance_dependencies``
- ``alternative_dependencies``
- ``dependency_modules``
"""
release_id = resolve_release_id(
self.session, release_id=release_id, release_code=release_code
)
empty_result: Dict[str, Any] = {
"intra_instance_validations": [],
"cross_instance_dependencies": [],
"alternative_dependencies": [],
"dependency_modules": {},
}
is_cross = scope_result.is_cross_module
# Normalised up front: a bare string left as-is would be
# iterated character by character downstream and manufacture
# periods like ``"-"`` and ``"1"`` out of ``"T-1Q"``.
ts = {
table: [periods] if isinstance(periods, str) else list(periods)
for table, periods in (time_shifts or {}).items()
}
# #325: a ``time_shift`` over a table of the reporting module
# needs another *instance of that same module*, so the operation
# is not intra-instance and the shifted instance has to be
# declared like any other cross-instance dependency — under the
# home module's own URI and the shift's reference period.
# Without this the shift is silently dropped: the engine resolves
# both operands against the current instance and the script fails.
home_periods = (
self._home_shift_ref_periods(
ts=ts,
primary_module_vid=primary_module_vid,
release_id=release_id,
home_module_tables=home_module_tables,
)
if not scope_result.has_error
else []
)
# Issue #120: when the primary module can evaluate the operation
# on its own (it appears as a single-module scope), prefer the
# intra-instance reading even if cross-instance scopes also exist
# for other modules. Only when the primary appears *solely* in
# multi-module scopes is it a genuine cross-instance dependency.
#
# A single-module scope isn't the only signal for that: a
# DB-native scope_result's OperationScope rows can omit a
# genuine single-module combination (dpmcore#364), so also check
# table ownership directly, mirroring mdpm's own is_intra_operation.
primary_has_intra = not scope_result.has_error and (
any(
{
c.module_vid
for c in getattr(s, "operation_scope_compositions", [])
}
== {primary_module_vid}
for s in scope_result.scopes or []
)
or bool(
referenced_tables
and home_module_tables is not None
and referenced_tables <= home_module_tables
)
)
if scope_result.has_error or not is_cross or primary_has_intra:
# This branch builds no dependency modules (the primary owns
# the reading, or the scope is not cross-module), so there are
# no genuine dependencies for two modules to be alternatives
# of: an empty valid-URI set drops every candidate pair (#202).
alternative_deps: List[List[str]] = []
if compute_alternative_deps and not scope_result.has_error:
alternative_deps = self.detect_alternative_dependencies(
scope_results=[scope_result],
primary_module_vid=primary_module_vid,
release_id=release_id,
valid_module_uris=set(),
)
# A shifted home table turns the intra claim into a
# cross-instance dependency on the home module itself (#325).
# The primary still has to own a single-module scope: a module
# hosting none of the referenced tables is neither the
# intra-instance owner nor the shifted instance's reporter.
self_deps = (
self._build_home_instance_deps(
periods=home_periods,
primary_module_vid=primary_module_vid,
operation_code=operation_code,
)
if primary_has_intra
else []
)
# The intra claim needs the primary to actually own a
# single-module scope. Keying it off ``not is_cross`` instead
# declared intra for a module that participates in no scope at
# all — it hosts none of the referenced tables (#141). That was
# masked while a redundant superset scope kept ``is_cross``
# true; dropping those scopes (#304) exposes it.
# ``not self_deps`` rather than ``not home_periods``: when the
# shifted instance cannot be declared (no resolvable URI) the
# operation keeps its intra classification instead of being
# dropped from the script's dependency block entirely.
return {
**empty_result,
"intra_instance_validations": (
[operation_code]
if operation_code and primary_has_intra and not self_deps
else []
),
"cross_instance_dependencies": self_deps,
"alternative_dependencies": alternative_deps,
}
valid_vids = self.filter_valid_dependency_modules(
scope_result, primary_module_vid
)
if not valid_vids:
# Scopes exist, but the primary module hosts none of the
# referenced tables and so participates in none of them: it is
# neither the intra-instance owner nor a cross-instance partner.
return {**empty_result}
# Build cross_instance_dependencies and dependency_modules
cross_deps: List[Dict[str, Any]] = []
dep_modules: Dict[str, Any] = {}
sorted_vids = sorted(valid_vids)
mv_rows = chunked_in(
self.session.query(ModuleVersion),
ModuleVersion.module_vid,
sorted_vids,
)
mv_by_vid = {mv.module_vid: mv for mv in mv_rows}
# Tables owned by the primary (home) module — a dep module that
# also lists any of these is sharing a table with the home; the
# sharing belongs to the home declaration, not to the dep, so
# exclude them from dependency_modules[<dep>].tables. Without
# this, cross-module ops that touch a shared table declare it
# twice (once in home, once in every dep that also owns it).
# Caller-supplied ``home_module_tables`` avoids the per-op
# recompute when this method runs inside a loop with a fixed
# ``primary_module_vid`` (the query is a per-table variable/
# open-key fetch, not a code-only lookup).
if home_module_tables is None:
primary_tables = set(
self._get_module_tables(
primary_module_vid, release_id=release_id
).keys()
)
else:
primary_tables = home_module_tables
home_mv, _ = self._home_mv_and_uri(primary_module_vid)
for vid in sorted_vids:
mv = mv_by_vid.get(vid)
if not mv:
continue
entry = self._build_dependency_entry(
vid=vid,
mv=mv,
release_id=release_id,
ts=ts,
operation_code=operation_code,
referenced_variables=referenced_variables,
referenced_tables=referenced_tables,
home_module_tables=primary_tables,
home_mv=home_mv,
)
if entry is None:
continue
entry_deps, uri, dep_module = entry
cross_deps.extend(entry_deps)
dep_modules[uri] = dep_module
# A cross-module operation can shift a home table too, and that
# shifted home instance is a dependency of its own (#325). It is
# appended after the external ones so their order — and the entry
# every existing caller reads first — stays unchanged.
cross_deps.extend(
self._build_home_instance_deps(
periods=home_periods,
primary_module_vid=primary_module_vid,
operation_code=operation_code,
)
)
alternative_deps = (
self.detect_alternative_dependencies(
scope_results=[scope_result],
primary_module_vid=primary_module_vid,
release_id=release_id,
valid_module_uris=set(dep_modules),
)
if compute_alternative_deps
else []
)
return {
"intra_instance_validations": [],
"cross_instance_dependencies": cross_deps,
"alternative_dependencies": alternative_deps,
"dependency_modules": dep_modules,
}
def _home_shift_ref_periods(
self,
ts: Dict[str, List[str]],
primary_module_vid: int,
release_id: Optional[int],
home_module_tables: Optional[Set[str]] = None,
) -> List[str]:
"""Return the reference periods shifted *within* the home module.
``ts`` maps table codes to the reference periods they are read
at. A table owned by the primary (home) module and read at a
period other than ``"T"`` means the operation reads another
instance of the reporting module itself (#325), which is a
cross-instance dependency the script has to declare — the
reference period never reached the output before, because it was
only ever read off a *dependency* module's tables.
One entry per distinct period, sorted for determinism: a single
operation shifting two home tables by different amounts needs
both instances, and the schema carries a period per declared
module rather than per table.
"""
shifted = {
tcode: [rp for rp in periods if rp and rp != "T"]
for tcode, periods in ts.items()
}
shifted = {tcode: rps for tcode, rps in shifted.items() if rps}
if not shifted:
return []
# Only pay for the table lookup once a shift is actually present.
tables = home_module_tables
if tables is None:
tables = set(
self._get_module_tables(
primary_module_vid, release_id=release_id
).keys()
)
return sorted(
{
rp
for tcode, rps in shifted.items()
if tcode in tables
for rp in rps
}
)
def _home_mv_and_uri(
self, primary_module_vid: int
) -> Tuple[Any, Optional[str]]:
"""Return (and cache) the home module's row and its URI.
Memoised in ``_home_module_refs``: callers iterate this once per
operation with the same ``primary_module_vid``.
"""
cached = self._home_module_refs.get(primary_module_vid)
if cached is None:
mv = (
self.session.query(ModuleVersion)
.filter(ModuleVersion.module_vid == primary_module_vid)
.first()
)
cached = (
mv,
self._get_module_uri(module_vid=primary_module_vid, mv=mv),
)
self._home_module_refs[primary_module_vid] = cached
return cached
def _build_home_instance_deps(
self,
periods: List[str],
primary_module_vid: int,
operation_code: Optional[str],
) -> List[Dict[str, Any]]:
"""Build the cross-instance entries for shifted home instances.
Each period yields one entry naming the home module's own URI —
the same URI the script is published under — so the engine knows
which instance to load for the shifted operand (#325). No
``dependency_modules`` entry accompanies it: the home module's
tables and variables are already declared at the top level of
the script, and the shifted instance is the same module version.
"""
if not periods:
return []
mv, uri = self._home_mv_and_uri(primary_module_vid)
if not uri:
return []
return [
self._cross_dep_entry(
uri=uri,
ref_period=period,
mv=mv,
operation_code=operation_code,
)
for period in periods
]
@staticmethod
def _cross_dep_entry(
uri: str,
ref_period: str,
mv: Any,
operation_code: Optional[str],
home_mv: Any = None,
) -> Dict[str, Any]:
"""Build one ``cross_instance_dependencies`` entry for *uri*.
``module_version`` is omitted when the row carries no version
number. The reference dates are the dependency module's own
window narrowed to the home module's (``home_mv``) window: the
two instances only coexist where both are active, so
``from_reference_date`` is the later of the two starts and
``to_reference_date`` the earlier of the two ends (an unset
bound on either side does not narrow). ``home_mv`` is omitted
for the home module's own shifted-instance entries (#325),
where ``mv`` already is the home module.
"""
module_entry: Dict[str, Any] = {
"URI": uri,
"ref_period": ref_period,
}
version_number = getattr(mv, "version_number", None)
if version_number:
module_entry["module_version"] = version_number
from_date = getattr(mv, "from_reference_date", None)
to_date = getattr(mv, "to_reference_date", None)
if home_mv is not None:
home_from = getattr(home_mv, "from_reference_date", None)
home_to = getattr(home_mv, "to_reference_date", None)
from_candidates = [d for d in (from_date, home_from) if d]
from_date = max(from_candidates) if from_candidates else None
to_candidates = [d for d in (to_date, home_to) if d]
to_date = min(to_candidates) if to_candidates else None
return {
"modules": [module_entry],
"affected_operations": (
[operation_code] if operation_code else []
),
"from_reference_date": (str(from_date) if from_date else ""),
"to_reference_date": (str(to_date) if to_date else ""),
}
def _build_dependency_entry(
self,
vid: int,
mv: Any,
release_id: Optional[int],
ts: Dict[str, List[str]],
operation_code: Optional[str],
referenced_variables: Optional[Dict[str, str]] = None,
referenced_tables: Optional[Set[str]] = None,
home_module_tables: Optional[Set[str]] = None,
home_mv: Any = None,
) -> Optional[Tuple[List[Dict[str, Any]], str, Dict[str, Any]]]:
"""Build the (cross_deps, uri, dependency_module) triple for *vid*.
One ``cross_instance_dependencies`` entry per distinct reference
period the module is read at — the schema carries a period per
declared module, so a module needed at two instances has to be
declared twice (#326). ``dependency_modules`` stays keyed by URI
and carries the union of the tables, exactly as before: the
table and datapoint definitions do not vary by instance.
``home_mv`` is the primary (home) module's own row, passed
through to :meth:`_cross_dep_entry` so each entry's reference
dates are narrowed to the window the two instances actually
share (#380) instead of the dependency's window alone.
Returns ``None`` when the module has no resolvable URI or
when every one of its tables is variable-less (and therefore
dropped, since the engine schema requires
``minProperties: 1`` on each table's variables map).
``home_module_tables`` is the set of table codes owned by the
primary (home) module. Tables that appear in both the home and
the dependency (a shared table like ``I_05.00`` present in both
IF_CLASS2 and IF_CLASS3) belong to the home declaration and
must be excluded from ``dependency_modules[<dep>].tables`` —
otherwise the engine sees the same table declared on both sides
and downstream lookups can pick the wrong copy.
"""
uri = self._get_module_uri(module_vid=vid, mv=mv)
if not uri:
return None
tables_dict_full = self._get_module_tables(vid, release_id=release_id)
tables_dict = {
tcode: tdata
for tcode, tdata in tables_dict_full.items()
if tdata.get("variables")
}
if not tables_dict:
return None
# The timeshift is a module-level property carried by the dependency
# module's tables. Compute it BEFORE narrowing: narrowing drops any
# table not referenced by the cross-rules, and a dropped table takes
# its timeshift with it — a module whose only timeshifted table is
# not referenced would otherwise fall back to ref_period T.
# Every period any of the module's tables is read at is kept:
# taking one dropped the others, so an operation reading two
# tables of the same module at different instances — or one
# table both plain and shifted — was declared against a single
# instance and evaluated against the wrong one (#326). Sorted so
# the output does not depend on visit order; ``"T"`` sorts first.
periods = sorted(
{
rp
for tbl_code in tables_dict
for rp in ts.get(tbl_code, ())
if rp
}
) or ["T"]
# #250: declare only the tables and datapoints the cross-rules
# actually reference — a whole dependency module is 100+ tables and
# 10k+ variables, where native EBA scripts declare a handful.
narrowed = self._narrow_dependency_tables(
tables_dict, referenced_tables, referenced_variables
)
if narrowed:
tables_dict = narrowed
# Drop tables the primary (home) module also declares — the shared
# ones belong to the home declaration; leaving them on the dep
# side is the duplicate-declaration bug this method exists to
# fix. Runs AFTER narrowing so a validation whose operand set is
# entirely inside a shared table still narrows correctly:
# excluding *before* narrowing would drop the operand's table
# first, narrowing would find no referenced table, and the
# narrowing fallback would reintroduce the shared table via the
# module-wide unnarrowed set. The exclusion is unconditional
# — even when it empties ``tables_dict``: the dep still surfaces
# in ``cross_instance_dependencies`` via its ``URI`` and the
# engine (#251) resolves cross-instance operands off
# ``referenced_variables`` when the caller supplies them.
if home_module_tables:
tables_dict = {
tcode: tdata
for tcode, tdata in tables_dict.items()
if tcode not in home_module_tables
}
cross_deps = [
self._cross_dep_entry(
uri=uri,
ref_period=period,
mv=mv,
operation_code=operation_code,
home_mv=home_mv,
)
for period in periods
]
variables: Dict[str, str] = {
k: v
for tbl in tables_dict.values()
for k, v in tbl.get("variables", {}).items()
}
# #251: the engine resolves *every* operand of a cross-instance
# validation against this map — including operands owned by the
# home module. A referenced datapoint missing here leaves the
# engine unable to build that operand: a bare single-cell home
# operand fails with "Scalar can't be created for this data" and
# an aggregated one silently computes 0. The dependency module's
# own definition of a datapoint wins over the referencing AST's.
for var_id, type_code in sorted((referenced_variables or {}).items()):
variables.setdefault(var_id, type_code)
dep_module = {
"tables": tables_dict,
"variables": variables,
}
return cross_deps, uri, dep_module
@staticmethod
def _narrow_dependency_tables(
tables_dict: Dict[str, Any],
referenced_tables: Optional[Set[str]],
referenced_variables: Optional[Dict[str, str]],
) -> Dict[str, Any]:
"""Restrict a dependency module to its referenced tables/datapoints.
Returns ``{}`` when the caller supplied no reference information,
or when narrowing would leave nothing declarable. The caller then
keeps the unnarrowed module: over-declaring is wrong, but dropping
a genuine cross-instance dependency outright is worse.
"""
if referenced_tables is None and referenced_variables is None:
return {}
narrowed: Dict[str, Any] = {}
for tcode, tdata in tables_dict.items():
if (
referenced_tables is not None
and tcode not in referenced_tables
):
continue
variables = tdata.get("variables", {})
if referenced_variables is not None:
variables = {
var_id: type_code
for var_id, type_code in variables.items()
if var_id in referenced_variables
}
# An empty variables map violates the engine schema's
# ``minProperties: 1``, so a table narrowed down to nothing is
# dropped rather than declared empty.
if not variables:
continue
narrowed[tcode] = {**tdata, "variables": variables}
return narrowed
# ------------------------------------------------------------------ #
# Alternative dependency detection (Fix 3)
# ------------------------------------------------------------------ #
[docs]
def detect_alternative_dependencies(
self,
scope_results: List[ScopeResult],
primary_module_vid: int,
release_id: Optional[int] = None,
release_code: Optional[str] = None,
valid_module_uris: Optional[Set[str]] = None,
) -> List[List[str]]:
"""Detect disjoint groups of interchangeable external modules.
Two external modules are alternatives only when they are
interchangeable dependencies of the *same* operation (#202):
within a single operation's scopes they each appear as the sole
external module alongside the primary, yet never co-exist in one
scope. Each ``ScopeResult`` is one operation, so candidate pairs
are collected per scope result — being the sole external of two
*different* operations does not make two modules alternatives.
Interchangeable pairs are then collapsed into disjoint groups so
that when three or more modules are mutually interchangeable they
surface as one group rather than every overlapping pair (#242).
Args:
scope_results: One entry per operation.
primary_module_vid: VID of the primary module.
release_id: Optional release filter (validated only).
release_code: Optional release code (validated only).
valid_module_uris: When given, the genuine dependency-module
URIs of the script. Pairs referencing a module outside
this set are dropped so ``alternative_dependencies`` can
never name a module absent from ``dependency_modules``
(#202 dangling references).
Returns a list of disjoint groups; each group is a sorted list
of two or more interchangeable module URIs, and no module appears
in more than one group.
"""
# Validate the release inputs (rejects an unknown code, or both
# arguments at once). The resolved id is not threaded further:
# each module version's URI roots at its own start release, not
# the report release.
resolve_release_id(
self.session, release_id=release_id, release_code=release_code
)
# Candidates: pairs sole-external within the *same* operation.
# Co-occurrence is tracked across every operation — two modules
# that ever share one scope are conjunctive, never alternatives.
candidate_pairs: Set[Tuple[int, int]] = set()
co_occurring: Set[Tuple[int, int]] = set()
for sr in scope_results:
single_ext_vids, ext_vid_sets = self._collect_external_vid_sets(
[sr], primary_module_vid
)
co_occurring |= self._co_occurring_pairs(ext_vid_sets)
candidate_pairs |= self._sole_external_pairs(single_ext_vids)
alt_pairs = sorted(candidate_pairs - co_occurring)
if not alt_pairs:
return []
uri_pairs = self._map_pairs_to_uris(alt_pairs)
if valid_module_uris is not None:
uri_pairs = [
pair
for pair in uri_pairs
if pair[0] in valid_module_uris
and pair[1] in valid_module_uris
]
# Interchangeable pairs overlap when 3+ modules are mutually
# interchangeable (A-B, A-C, B-C). Collapse them into the disjoint
# groups the consumer expects: one connected component == one
# dependency slot fillable by any module in it (#242).
return self._group_alternative_pairs(uri_pairs)
@staticmethod
def _collect_external_vid_sets(
scope_results: List[ScopeResult],
primary_module_vid: int,
) -> tuple[Set[int], List[frozenset[int]]]:
"""Extract external VID sets from scopes."""
single_ext_vids: Set[int] = set()
all_ext_vid_sets: List[frozenset[int]] = []
for sr in scope_results:
for scope in sr.scopes or []:
scope_vids = {
c.module_vid
for c in getattr(
scope,
"operation_scope_compositions",
[],
)
}
if primary_module_vid not in scope_vids or len(scope_vids) < 2:
continue
ext_vids = frozenset(scope_vids - {primary_module_vid})
all_ext_vid_sets.append(ext_vids)
if len(ext_vids) == 1:
single_ext_vids.update(ext_vids)
return single_ext_vids, all_ext_vid_sets
@staticmethod
def _sole_external_pairs(
single_ext_vids: Set[int],
) -> Set[tuple[int, int]]:
"""All sorted VID pairs among sole-external modules of one op."""
sorted_vids = sorted(single_ext_vids)
return {
(v1, v2)
for i, v1 in enumerate(sorted_vids)
for v2 in sorted_vids[i + 1 :]
}
@staticmethod
def _co_occurring_pairs(
all_ext_vid_sets: List[frozenset[int]],
) -> Set[tuple[int, int]]:
"""Sorted VID pairs that share a single scope (conjunctive)."""
co_occurring: Set[tuple[int, int]] = set()
for ext_set in all_ext_vid_sets:
if len(ext_set) > 1:
sorted_vids = sorted(ext_set)
for i, v1 in enumerate(sorted_vids):
for v2 in sorted_vids[i + 1 :]:
co_occurring.add((v1, v2))
return co_occurring
@staticmethod
def _group_alternative_pairs(
uri_pairs: List[List[str]],
) -> List[List[str]]:
"""Collapse overlapping interchangeable pairs into disjoint groups.
Nodes are module URIs, edges are interchangeable pairs; each
connected component becomes one group. Guarantees the returned
groups share no module (disjoint), which is the shape the
consuming engine's dependency report expects (#242).
"""
adjacency: Dict[str, Set[str]] = {}
for a, b in uri_pairs:
adjacency.setdefault(a, set()).add(b)
adjacency.setdefault(b, set()).add(a)
seen: Set[str] = set()
groups: List[List[str]] = []
for start in sorted(adjacency):
if start in seen:
continue
stack, component = [start], set()
while stack:
node = stack.pop()
if node in seen:
continue
seen.add(node)
component.add(node)
stack.extend(adjacency[node] - seen)
groups.append(sorted(component))
return sorted(groups)
def _map_pairs_to_uris(
self,
pairs: List[tuple[int, int]],
) -> List[List[str]]:
"""Resolve VID pairs to sorted URI pairs."""
needed: Set[int] = set()
for v1, v2 in pairs:
needed.add(v1)
needed.add(v2)
mv_by_vid: Dict[int, Any] = {}
if needed:
mv_rows = chunked_in(
self.session.query(ModuleVersion),
ModuleVersion.module_vid,
needed,
)
mv_by_vid = {mv.module_vid: mv for mv in mv_rows}
vid_to_uri: Dict[int, str] = {}
for vid in needed:
uri = self._get_module_uri(
module_vid=vid,
mv=mv_by_vid.get(vid),
)
if uri:
vid_to_uri[vid] = uri
result: List[List[str]] = []
for v1, v2 in pairs:
uri1 = vid_to_uri.get(v1)
uri2 = vid_to_uri.get(v2)
if uri1 and uri2:
result.append(sorted([uri1, uri2]))
return result
# ------------------------------------------------------------------ #
# Helpers
# ------------------------------------------------------------------ #
def _get_module_tables(
self,
module_vid: int,
release_id: Optional[int] = None,
ghost_vids: Optional[List[int]] = None,
) -> Dict[str, Any]:
"""Return tables for a module with their variables and open keys.
Returns::
{table_code: {"variables": {var_id: type_code},
"open_keys": {property_code: data_type_code}}}
``release_id`` filters the open-keys query by release window.
``ghost_vids`` (see :meth:`_ghost_chain_vids`) additionally
unions in the TableVersion/variable rows composed against those
#182 ghost module versions. A ghost-rescued operation (see
:meth:`_substitute_ghost_compositions`) can reference a
variable that only exists on the ghost's own ``TableVersion``,
never composed against the fallback's ``module_vid`` — leaving
it out of the declared variables map makes the downstream
engine unable to build a Scalar for that operand ("Scalar
can't be created for this data").
"""
# Get table codes + VIDs for this module (plus any ghost siblings
# whose own table/variable rows the fallback doesn't compose).
all_vids = [module_vid, *(ghost_vids or [])]
tv_rows = (
self.session.query(
TableVersion.code,
TableVersion.table_vid,
)
.join(
ModuleVersionComposition,
TableVersion.table_vid == ModuleVersionComposition.table_vid,
)
.filter(ModuleVersionComposition.module_vid.in_(all_vids))
.all()
)
table_vids = [r.table_vid for r in tv_rows if r.table_vid]
vid_to_code = {r.table_vid: r.code for r in tv_rows if r.code}
# Batch-fetch variables for all tables at once
variables_by_tvid: Dict[int, Dict[str, str]] = {
tvid: {} for tvid in table_vids
}
if table_vids:
var_base = (
self.session.query(
TableVersionCell.table_vid,
Variable.variable_id,
DataType.code,
)
.select_from(TableVersionCell)
.join(
VariableVersion,
TableVersionCell.variable_vid
== VariableVersion.variable_vid,
)
.join(
Variable,
VariableVersion.variable_id == Variable.variable_id,
)
.join(
Property,
VariableVersion.property_id == Property.property_id,
)
.join(
DataType,
Property.data_type_id == DataType.data_type_id,
)
.distinct()
)
var_rows = chunked_in(
var_base, TableVersionCell.table_vid, table_vids
)
for row in var_rows:
tvid = row[0]
var_id = str(row[1])
type_code = row[2] or ""
if tvid in variables_by_tvid:
variables_by_tvid[tvid][var_id] = type_code
# Open keys per table_code, pinned to this module version's own
# TableVersion rows (not just their codes — see _open_keys.py).
# A ghost sibling's TableVersion can share a code with the
# fallback's own (see ``ghost_vids`` above), so dedupe codes.
open_keys_by_code = _get_open_keys_for_tables(
self.session,
sorted(set(vid_to_code.values())),
release_id=release_id,
table_vids=table_vids,
)
# Union rather than overwrite: a ghost's TableVersion and the
# fallback's own can share a code (same conceptual table under
# two different rows), each with its own, disjoint variable_ids.
tables: Dict[str, Any] = {}
for tvid, code in vid_to_code.items():
entry = tables.setdefault(
code,
{
"variables": {},
"open_keys": open_keys_by_code.get(code, {}),
},
)
entry["variables"].update(variables_by_tvid.get(tvid, {}))
return tables
def _get_module_uri(
self,
module_vid: int,
mv: Optional[Any] = None,
) -> Optional[str]:
"""Resolve a module VID to its EBA taxonomy URI.
The URI's release segment always comes from the release in which
the module version was introduced (its start release), resolved
via :meth:`_resolve_uri_release_id`. A module version's taxonomy
is published under that release, so an unchanged module keeps its
original release segment even inside a later report: e.g. an
unchanged ``ae`` stays at ``.../ae/4.2/mod/ae`` in a 4.2.1 report,
because no ``ae`` taxonomy exists at 4.2.1. The report release is
deliberately not an input: it never sets the release segment, and
it would not pick the module version either — the lookup filters
by ``module_vid`` alone.
Resolution order (in :meth:`_resolve_uri_release_id`): the static
CSV mapping by ``(module_code, version_number)`` first; on a miss,
the module version's ``start_release_id``.
When *mv* is supplied the initial DB lookup is skipped.
"""
try:
if mv is None:
mv = (
self.session.query(ModuleVersion)
.filter(ModuleVersion.module_vid == module_vid)
.first()
)
if not mv or not mv.module:
return None
module_code = mv.code
if not module_code:
return None
framework = mv.module.framework
if not framework or not framework.code:
return None
csv_or_release_id = self._resolve_uri_release_id(mv, module_code)
if isinstance(csv_or_release_id, str):
return csv_or_release_id # CSV hit (already final URI).
if csv_or_release_id is None:
return None
release_row = (
self.session.query(Release.code)
.filter(Release.release_id == csv_or_release_id)
.first()
)
if not release_row or not release_row.code:
return None
return (
"http://www.eba.europa.eu/eu/fr/xbrl/crr"
"/fws/"
f"{framework.code.lower()}/"
f"{release_row.code}/mod/"
f"{module_code.lower()}"
)
except Exception as exc:
logger.warning(
"Failed to resolve URI for module VID %s: %s",
module_vid,
exc,
)
return None
@staticmethod
def _resolve_uri_release_id(
mv: Any,
module_code: str,
) -> Optional[Any]:
"""Pick the release that seeds the URL's release segment.
A module version's taxonomy is published under the release in
which that version was introduced (its ``start_release_id``), not
under the report release. Seeding the segment from the report
release would build URIs like ``.../ae/4.2.1/mod/ae`` for modules
that did not change since 4.2 and therefore have no taxonomy at
4.2.1. So the segment is resolved from the module version itself,
the same way for every caller.
Returns one of:
- ``str`` — a final URI (CSV hit, ``.json`` suffix already
stripped). The caller must short-circuit and return it.
- ``int`` — the release_id whose ``Release.code`` should fill
the ``/{release}/`` segment.
- ``None`` — nothing resolvable; caller returns ``None``.
"""
from dpmcore.data import (
get_module_schema_ref_by_version,
)
if mv.version_number:
static = get_module_schema_ref_by_version(
module_code, mv.version_number
)
if static:
return static.removesuffix(".json")
return mv.start_release_id