"""Unified DPM-XL service facade."""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional, TypedDict
from dpmcore.services.ast_generator import ASTGeneratorService
from dpmcore.services.scope_calculator import ScopeCalculatorService
from dpmcore.services.semantic import ParameterInfo, SemanticService
from dpmcore.services.syntax import SyntaxService
if TYPE_CHECKING:
from sqlalchemy.orm import Session
class SyntaxValidationResult(TypedDict):
"""Shape of ``DpmXlService.validate_syntax`` return value."""
is_valid: bool
error_message: str | None
expression: str
class SemanticValidationResult(TypedDict):
"""Shape of ``DpmXlService.validate_semantic`` return value.
``is_valid`` is pair-wide when a precondition expression was supplied, and
``error_source`` then names the failing half. See
:class:`~dpmcore.services.semantic.SemanticResult`.
"""
is_valid: bool
error_message: str | None
error_code: str | None
expression: str
warning: str | None
error_source: str | None
[docs]
class DpmXlService:
"""One-stop facade for all DPM-XL operations.
Composes the individual services and provides convenient shortcuts
for common workflows.
Args:
session: An open SQLAlchemy session (optional for syntax-only
operations).
"""
[docs]
def __init__(self, session: Optional["Session"] = None) -> None:
"""Build the facade, optionally bound to a SQLAlchemy ``session``."""
self.syntax = SyntaxService()
self.session = session
if session is not None:
self.semantic = SemanticService(session)
self.ast_generator = ASTGeneratorService(session)
self.scope_calculator = ScopeCalculatorService(session)
else:
self.semantic = None # type: ignore[assignment]
self.ast_generator = ASTGeneratorService()
self.scope_calculator = None # type: ignore[assignment]
[docs]
def validate_syntax(self, expression: str) -> SyntaxValidationResult:
"""Validate syntax only (no DB required)."""
result = self.syntax.validate(expression)
return {
"is_valid": result.is_valid,
"error_message": result.error_message,
"expression": result.expression,
}
[docs]
def validate_semantic(
self,
expression: str,
release_id: Optional[int] = None,
release_code: Optional[str] = None,
*,
precondition_expression: Optional[str] = None,
) -> SemanticValidationResult:
"""Full semantic validation, optionally gated (requires DB).
``precondition_expression`` is keyword-only and appended after the
pre-existing arguments, so ``validate_semantic(expr, 5)`` still means
``release_id=5``.
"""
if self.semantic is None:
raise RuntimeError("No database session provided.")
result = self.semantic.validate(
expression,
precondition_expression=precondition_expression,
release_id=release_id,
release_code=release_code,
)
return {
"is_valid": result.is_valid,
"error_message": result.error_message,
"error_code": result.error_code,
"expression": result.expression,
"warning": result.warning,
"error_source": result.error_source,
}
[docs]
def get_parameters(
self,
expression: str,
release_id: Optional[int] = None,
release_code: Optional[str] = None,
) -> tuple[ParameterInfo, ...]:
"""Return the parameters referenced by *expression* (requires DB).
Surfaces the runtime-binding contract: each referenced parameter with
its declared type, set-ness and default. Returns an empty tuple when
the expression references no parameters or fails validation.
"""
if self.semantic is None:
raise RuntimeError("No database session provided.")
result = self.semantic.validate(
expression,
release_id=release_id,
release_code=release_code,
)
return result.parameters