"""Operation scope calculation service."""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Optional,
Set,
Tuple,
)
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.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__)
[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()
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),
)
# ------------------------------------------------------------------ #
# 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[Dict[str, 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
ref-period strings (e.g. ``{"C_01.00": "T-1Q"}``).
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
ts = time_shifts or {}
# 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.
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 []
)
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(),
)
# 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.
return {
**empty_result,
"intra_instance_validations": (
[operation_code]
if operation_code and primary_has_intra
else []
),
"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
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,
)
if entry is None:
continue
cross_dep, uri, dep_module = entry
cross_deps.append(cross_dep)
dep_modules[uri] = dep_module
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 _build_dependency_entry(
self,
vid: int,
mv: Any,
release_id: Optional[int],
ts: Dict[str, 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,
) -> Optional[Tuple[Dict[str, Any], str, Dict[str, Any]]]:
"""Build a single (cross_dep, uri, dependency_module) triple.
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.
ref_period = "T"
for tbl_code in tables_dict:
rp = ts.get(tbl_code)
if rp and rp != "T":
ref_period = rp
# #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
}
module_entry: Dict[str, Any] = {
"URI": uri,
"ref_period": ref_period,
}
if mv.version_number:
module_entry["module_version"] = mv.version_number
from_date = mv.from_reference_date
to_date = mv.to_reference_date
cross_dep = {
"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 ""),
}
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_dep, 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,
) -> 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.
"""
# Get table codes + VIDs for this module
tv_rows = (
self.session.query(
TableVersion.code,
TableVersion.table_vid,
)
.join(
ModuleVersionComposition,
TableVersion.table_vid == ModuleVersionComposition.table_vid,
)
.filter(ModuleVersionComposition.module_vid == module_vid)
.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
open_keys_by_code = _get_open_keys_for_tables(
self.session,
list(vid_to_code.values()),
release_id=release_id,
)
tables: Dict[str, Any] = {}
for tvid, code in vid_to_code.items():
tables[code] = {
"variables": variables_by_tvid.get(tvid, {}),
"open_keys": open_keys_by_code.get(code, {}),
}
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