class ModelStore:
"""In-memory model registry. Thread-safe via ``threading.Lock``.
Models are keyed by a 16-char hex id: content-derived for shared models
(see ``service/model_cache.py``) or random for private ones. All parsing,
validation, and compilation infrastructure is instantiated internally,
following the same singleton pattern as ``api/deps.py``.
"""
def __init__(
self,
max_models: int = 10,
shared_cache: ModelCache | None = None,
compilation_cache: CompilationCache | None = None,
) -> None:
self._lock = threading.Lock()
# Process-wide compilation cache; None compiles every time (CLI,
# stateless helpers, bare unit tests).
self._compilation_cache = compilation_cache
# Process-wide content-addressed cache shared across sessions. When
# None (CLI, stateless helpers, bare unit tests) every model stays
# private to this store and ids are random — behaviour identical to
# before the cache existed.
self._shared_cache = shared_cache
# model_ids in this store that are backed by ``_shared_cache`` (so
# remove_model/close release the shared reference).
self._shared_ids: set[str] = set()
self._models: dict[str, SemanticModel] = {}
# Parallel storage of each loaded model's *merged* raw YAML dict
# so inheritance can re-merge against the exact same content the
# parent was built from. Pre-fix (v2.7.5) inheritance round-tripped
# through ``_model_to_raw`` which dropped most non-essential
# fields (numClass, primaryKey, expression on computed columns,
# measure dataType / filters / grain / delimiter / withinGroup,
# most metric subtype config, …) — child models would inherit
# a stripped parent and silently compile invalid SQL such as
# ``SUM("T"."")`` for any parent computed column whose ``code:``
# the resolver had derived from its ``expression``.
self._raws: dict[str, dict[str, object]] = {}
self._graphs: dict[str, GraphArtifact] = {}
# Per-store summary cache so dedup hits can return the original
# data_objects/dimensions/measures/metrics counts without re-walking
# the model.
self._summaries: dict[str, ModelSummary] = {}
self._max_models = max_models
# Dedup index: content_hash → model_id. Populated on every successful
# load and consulted before parsing on the next load. See
# design/PLAN_model_load_dedup.md.
self._content_hash_index: dict[str, str] = {}
# Internal pipeline singletons (stateless, safe to share).
self._loader = TrackedLoader()
self._resolver = ReferenceResolver()
self._validator = SemanticValidator()
self._merger = ExtendsMerger()
self._pipeline = CompilationPipeline()
# -- helpers -------------------------------------------------------------
@staticmethod
def _new_id() -> str:
return uuid.uuid4().hex[:16]
@staticmethod
def _content_id(content_hash: str) -> str:
"""Stable content-derived id for a shared model.
A 64-bit prefix of the content hash: identical OBML always yields the
same id (so sessions loading matching bytes collapse to one shared
compiled model and one result-cache key), while the full 64-hex
``content_hash`` remains the sharing key, so an id-label collision can
never cause a wrong-model share.
"""
return content_hash[:16]
def _register_shared(self, entry: CompiledModel) -> None:
"""Register a shared cache entry into this store's local view.
Caller must hold ``self._lock``. The store's read-through getters
(get_model/get_raw/get_graph/describe/list_models/...) then serve the
shared entry with no further changes.
"""
self._models[entry.model_id] = entry.model
self._raws[entry.model_id] = entry.raw
self._graphs[entry.model_id] = entry.graph
self._summaries[entry.model_id] = entry.summary
self._content_hash_index[entry.content_hash] = entry.model_id
self._shared_ids.add(entry.model_id)
def _adopt_shared(self, entry: CompiledModel) -> None:
"""Take ownership of one reference to ``entry`` for this store.
``entry``'s refcount was already incremented by the caller (via
``acquire`` / ``insert_or_acquire``). This store holds exactly one
reference per shared model_id it exposes, so a surplus reference (the
store already references this content, e.g. concurrent identical loads)
is released. Raises ``ModelCapacityError`` (releasing the reference
first) when the store is full.
"""
assert self._shared_cache is not None # only called on shared paths
with self._lock:
if entry.model_id in self._shared_ids:
surplus, over_cap = True, False
else:
surplus = False
over_cap = len(self._models) >= self._max_models
if not surplus and not over_cap:
self._register_shared(entry)
if surplus or over_cap:
# Release outside the store lock — the cache has its own lock.
self._shared_cache.release(entry.model_id)
if over_cap:
raise ModelCapacityError(f"Maximum models per session reached ({self._max_models})")
@staticmethod
def _health_for(model: SemanticModel) -> ModelHealthSummary:
"""Compute structural health for a loaded model."""
h = compute_health(model)
return ModelHealthSummary(
status=h.status,
data_objects=h.data_objects,
joins=h.joins,
orphan_data_objects=h.orphan_data_objects,
fan_trap_risks=[
FanTrapRiskInfo(
tables=r.tables,
reason=r.reason,
suggested_pattern=r.suggested_pattern,
)
for r in h.fan_trap_risks
],
unreachable_dimensions=h.unreachable_dimensions,
warnings_count=h.warnings_count,
)
@staticmethod
def _check_named_source(model: SemanticModel, execution_dialect: str | None) -> None:
"""Refuse a model already compiled once whose data source cannot run it.
A reused or shared model skips parsing, but not this: it may have been
loaded where nothing executes, or before the environment changed.
"""
if execution_dialect is not None:
errors = _named_source_errors(model, execution_dialect)
if errors:
raise ModelValidationError(errors, [])
@staticmethod
def _content_hash(yaml_str: str, datasource: str | None = None) -> str:
"""SHA-256 of the OBML body, with surrounding whitespace stripped.
Stripping at the boundary makes a trailing newline difference
invisible to dedup; everything else (key order, comments, internal
whitespace) still produces a different hash. A data source given at
load is part of the content: the same YAML on another source is
another model, with its own id and cache entries.
"""
content = yaml_str.strip()
if datasource is not None:
content = f"{content}\0defaultDataSource={datasource}"
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def _parse_and_validate(
self,
yaml_str: str | None = None,
*,
raw_dict: dict[str, object] | None = None,
extends_yaml: list[str] | None = None,
inherits_model_id: str | None = None,
datasource: str | None = None,
) -> tuple[SemanticModel, dict[str, object], list[ErrorInfo], list[ErrorInfo]]:
"""Parse YAML (or accept pre-parsed dict), resolve references, validate.
Returns ``(model, merged_raw, errors, warnings)``.
Provide either ``yaml_str`` or ``raw_dict``, not both. ``datasource``
replaces the model's ``settings.defaultDataSource``.
``merged_raw`` is the fully-merged raw dict the resolver consumed
(after extends/inherits processing) — callers store it so future
inherits-from-this-model loads can re-merge against the exact
content rather than going through a lossy ``_model_to_raw``
round-trip.
"""
errors: list[ErrorInfo] = []
warnings: list[ErrorInfo] = []
# 1. Parse YAML or use pre-parsed dict
if raw_dict is not None:
raw = raw_dict
source_map = None
elif yaml_str is not None:
try:
raw, source_map = self._loader.load_string(yaml_str)
except YAMLSafetyError as exc:
errors.append(ErrorInfo(code="YAML_SAFETY_ERROR", message=str(exc)))
return SemanticModel(), {}, errors, warnings
except Exception as exc:
errors.append(ErrorInfo(code="YAML_PARSE_ERROR", message=str(exc)))
return SemanticModel(), {}, errors, warnings
else:
errors.append(
ErrorInfo(
code="NO_MODEL_INPUT",
message="Provide either model_yaml or model_json",
)
)
return SemanticModel(), {}, errors, warnings
# 1b. Merge extends/inherits if provided
try:
inherits_raw: dict[str, object] | None = None
if inherits_model_id is not None:
# Prefer the parent's stored raw dict — captured at load
# time so every field round-trips intact. Fall back to
# the lossy ``_model_to_raw`` only when no raw is on
# record (legacy / programmatically-constructed models).
with self._lock:
inherits_raw = self._raws.get(inherits_model_id)
if inherits_raw is None:
parent_model = self.get_model(inherits_model_id)
inherits_raw = self._model_to_raw(parent_model)
if extends_yaml or inherits_raw is not None:
raw, merge_warnings = self._merger.merge_from_strings(
raw,
extend_yamls=extends_yaml,
inherits_raw=inherits_raw,
)
for mw in merge_warnings:
warnings.append(
ErrorInfo(
code=WarningCode.MERGE_WARNING,
message=mw,
severity="warning",
)
)
source_map = None
except MergeError as exc:
errors.append(ErrorInfo(code=exc.code, message=exc.message))
return SemanticModel(), {}, errors, warnings
except KeyError:
errors.append(
ErrorInfo(
code="PARENT_MODEL_NOT_FOUND",
message=f"Parent model '{inherits_model_id}' not found in session",
)
)
return SemanticModel(), {}, errors, warnings
# 1c. A data source chosen at load replaces the one the model names.
settings = raw.get("settings")
if datasource is not None and (settings is None or isinstance(settings, dict)):
raw = {**raw, "settings": {**(settings or {}), "defaultDataSource": datasource}}
# 2. Resolve references
model, resolution = self._resolver.resolve(raw, source_map)
for e in resolution.errors:
errors.append(
ErrorInfo(
code=e.code,
message=e.message,
path=e.path,
suggestions=list(e.suggestions),
severity=e.severity,
hint=e.hint,
context=e.context,
)
)
for w in resolution.warnings:
warnings.append(
ErrorInfo(
code=w.code,
message=w.message,
path=w.path,
suggestions=list(w.suggestions),
severity=w.severity or "warning",
hint=w.hint,
context=w.context,
)
)
# 3. Semantic validation. A few checks overlap the resolver's (an
# unknown data object, say); the resolver's report carries the
# suggestions, so the same issue is not listed twice.
reported = {(i.code, i.path, i.message) for i in (*errors, *warnings)}
sem_errors = self._validator.validate(model)
for e in sem_errors:
if (e.code, e.path, e.message) in reported:
continue
info = ErrorInfo(
code=e.code,
message=e.message,
path=e.path,
suggestions=list(e.suggestions),
severity=e.severity,
hint=e.hint,
context=e.context,
)
if e.severity == "warning":
warnings.append(info)
else:
errors.append(info)
# 4. Cross-dataObject refresh contract consistency check.
from orionbelt.cache.contracts import collect_table_contracts
_, refresh_warnings = collect_table_contracts(model)
for w in refresh_warnings:
warnings.append(
ErrorInfo(
code=w.code,
message=w.message,
path=w.path,
suggestions=list(w.suggestions),
severity=w.severity or "warning",
hint=w.hint,
context=w.context,
)
)
return model, raw, errors, warnings
@staticmethod
def _model_to_raw(model: SemanticModel) -> dict[str, object]:
"""Convert a SemanticModel back to a raw dict for inherits merging.
.. deprecated:: v2.7.5
Lossy fallback only — drops most non-essential fields. New
code stores and reuses the merged raw dict captured at load
time (see ``ModelStore._raws``). This method remains for the
edge case where a parent model was constructed programmatically
without ever passing through ``load_model``.
"""
raw: dict[str, object] = {"version": model.version}
if model.description:
raw["description"] = model.description
if not model.expose_counts:
raw["exposeCounts"] = False
if model.count_label_pattern != DEFAULT_COUNT_PATTERN:
raw["countLabelPattern"] = model.count_label_pattern
if model.data_objects:
objs: dict[str, object] = {}
for name, obj in model.data_objects.items():
obj_raw: dict[str, object] = {
"code": obj.code,
"database": obj.database,
"schema": obj.schema_name,
}
if obj.columns:
cols: dict[str, object] = {}
for cname, col in obj.columns.items():
cols[cname] = {
"code": col.code,
"abstractType": col.abstract_type.value,
}
obj_raw["columns"] = cols
if obj.joins:
joins: list[dict[str, object]] = []
for j in obj.joins:
jd: dict[str, object] = {
"joinType": j.join_type.value,
"joinTo": j.join_to,
"columnsFrom": list(j.columns_from),
"columnsTo": list(j.columns_to),
}
if j.secondary:
jd["secondary"] = True
jd["pathName"] = j.path_name
joins.append(jd)
obj_raw["joins"] = joins
# Carry count-synthesis knobs when non-default so inheritance
# does not silently re-enable a suppressed count.
if obj.data_source is not None:
obj_raw["dataSource"] = obj.data_source
if not obj.countable:
obj_raw["countable"] = False
if obj.count_label is not None:
obj_raw["countLabel"] = obj.count_label
if obj.refresh is not None:
refresh: dict[str, object] = {"mode": obj.refresh.mode}
if obj.refresh.interval:
refresh["interval"] = obj.refresh.interval
if obj.refresh.anchor:
refresh["anchor"] = obj.refresh.anchor
if obj.refresh.timezone:
refresh["timezone"] = obj.refresh.timezone
if obj.refresh.max_staleness:
refresh["maxStaleness"] = obj.refresh.max_staleness
obj_raw["refresh"] = refresh
objs[name] = obj_raw
raw["dataObjects"] = objs
if model.dimensions:
dims: dict[str, object] = {}
for name, dim in model.dimensions.items():
dd: dict[str, object] = {
"dataObject": dim.view,
"column": dim.column,
"resultType": dim.result_type.value,
}
if dim.time_grain:
dd["timeGrain"] = dim.time_grain.value
# Carried when off, so an inheriting model does not list the
# values of a dimension its parent keeps from being enumerated.
if not dim.value_search:
dd["valueSearch"] = False
dims[name] = dd
raw["dimensions"] = dims
if model.measures:
meas: dict[str, object] = {}
for name, m in model.measures.items():
md: dict[str, object] = {
"aggregation": m.aggregation,
"resultType": m.result_type.value,
}
if m.expression:
md["expression"] = m.expression
if m.columns:
md["columns"] = [
{"dataObject": c.view or "", "column": c.column or ""} for c in m.columns
]
if m.total:
md["total"] = True
meas[name] = md
raw["measures"] = meas
if model.metrics:
mets: dict[str, object] = {}
for name, met in model.metrics.items():
mtd: dict[str, object] = {"type": met.type.value}
if met.expression:
mtd["expression"] = met.expression
if met.measure:
mtd["measure"] = met.measure
if met.time_dimension:
mtd["timeDimension"] = met.time_dimension
mets[name] = mtd
raw["metrics"] = mets
if model.filters:
raw["filters"] = [
{
"dataObject": f.data_object,
"column": f.column,
"operator": f.operator,
**({"value": f.value} if f.value is not None else {}),
**({"values": f.values} if f.values else {}),
# A parent rebuilt for ``inherits`` must keep it: dropped,
# the child's tenant filter would fail open.
**(
{"enforce": f.enforce.value}
if f.enforce != FilterEnforcement.RELEVANT
else {}
),
}
for f in model.filters
]
return raw
# -- public API ----------------------------------------------------------
def load_model(
self,
yaml_str: str | None = None,
*,
raw_dict: dict[str, object] | None = None,
extends_yaml: list[str] | None = None,
inherits_model_id: str | None = None,
dedup: bool = True,
datasource: str | None = None,
execution_dialect: str | None = None,
) -> LoadResult:
"""Parse, validate, and store a model. Returns id + summary.
Provide either ``yaml_str`` or ``raw_dict``.
Raises ``ModelValidationError`` if the model has validation errors.
Raises ``ModelCapacityError`` if the session's model cap is reached.
When ``dedup`` is True (default) and the same OBML bytes have already
been loaded into this store, the existing ``model_id`` is returned
and ``model_load`` is set to ``"reused"``. Dedup only applies to
plain ``yaml_str`` loads — when ``raw_dict``, ``extends_yaml``, or
``inherits_model_id`` is supplied the load always runs fresh, since
the effective content depends on inputs not captured by the YAML
bytes alone.
``datasource`` names the data source the model's queries run on,
replacing its ``settings.defaultDataSource``. ``execution_dialect``,
given where queries execute, is the dialect of a model that sets no
``defaultDialect``; a data source the model names must then be
configured, and of the model's dialect, or the load fails.
"""
# Dedup is meaningful only for a stand-alone YAML body. The other
# input shapes either skip the YAML stage (raw_dict) or fold in
# additional state (extends/inherits) that the bytes don't capture.
dedup_eligible = (
dedup
and yaml_str is not None
and raw_dict is None
and not extends_yaml
and inherits_model_id is None
)
content_hash: str | None = None
if dedup_eligible:
content_hash = self._content_hash(yaml_str or "", datasource)
with self._lock:
existing_id = self._content_hash_index.get(content_hash)
if existing_id is not None and existing_id in self._models:
summary = self._summaries.get(existing_id)
if summary is not None:
existing_model = self._models[existing_id]
self._check_named_source(existing_model, execution_dialect)
existing_health = self._health_for(existing_model)
return LoadResult(
model_id=existing_id,
data_objects=summary.data_objects,
dimensions=summary.dimensions,
measures=summary.measures,
metrics=summary.metrics,
warnings=[],
model_load="reused",
health=existing_health,
)
# Stale index entry — drop it and fall through to a fresh load.
if existing_id is not None:
self._content_hash_index.pop(content_hash, None)
# Cross-session hit: another session already compiled these exact
# bytes. Adopt the shared compiled model with no recompile. From this
# store's perspective the load is still "fresh" (a new reference for
# this session); the compile-skip is a transparent optimisation.
if dedup_eligible and self._shared_cache is not None and content_hash is not None:
shared = self._shared_cache.acquire(content_hash)
if shared is not None:
try:
self._check_named_source(shared.model, execution_dialect)
except ModelValidationError:
self._shared_cache.release(shared.model_id)
raise
self._adopt_shared(shared)
return LoadResult(
model_id=shared.model_id,
data_objects=shared.summary.data_objects,
dimensions=shared.summary.dimensions,
measures=shared.summary.measures,
metrics=shared.summary.metrics,
warnings=[],
model_load="fresh",
health=self._health_for(shared.model),
)
with self._lock:
if len(self._models) >= self._max_models:
raise ModelCapacityError(f"Maximum models per session reached ({self._max_models})")
model, merged_raw, errors, warnings = self._parse_and_validate(
yaml_str,
raw_dict=raw_dict,
extends_yaml=extends_yaml,
inherits_model_id=inherits_model_id,
datasource=datasource,
)
if not errors and execution_dialect is not None:
errors = _named_source_errors(model, execution_dialect)
if errors:
raise ModelValidationError(errors, warnings)
shared_load = dedup_eligible and self._shared_cache is not None and content_hash is not None
# Content-derived id for shared models so identical bytes collapse to
# one id (and one result-cache key) across sessions; random id
# otherwise, preserving the ``dedup=False`` "distinct model" contract.
model_id = (
self._content_id(content_hash)
if content_hash is not None and shared_load
else self._new_id()
)
# Eagerly export OBSL-Core graph (Option C: at model load time).
graph = export_obsl(model, model_id)
turtle = graph.serialize(format="turtle")
artifact = GraphArtifact(graph=graph, turtle=turtle, generated_at=time.monotonic())
summary = ModelSummary(
model_id=model_id,
data_objects=len(model.data_objects),
dimensions=len(model.dimensions),
measures=len(model.measures),
metrics=len(model.metrics),
)
if shared_load:
assert self._shared_cache is not None and content_hash is not None
compiled = CompiledModel(
model_id=model_id,
content_hash=content_hash,
model=model,
raw=merged_raw,
graph=artifact,
summary=summary,
)
# A concurrent load of the same content may have won the race;
# insert_or_acquire returns the surviving entry either way.
entry = self._shared_cache.insert_or_acquire(compiled)
self._adopt_shared(entry)
return LoadResult(
model_id=entry.model_id,
data_objects=entry.summary.data_objects,
dimensions=entry.summary.dimensions,
measures=entry.summary.measures,
metrics=entry.summary.metrics,
warnings=warnings,
model_load="fresh",
health=self._health_for(entry.model),
)
with self._lock:
# Re-check capacity under lock — the first check (above) ran
# outside the lock while parsing/exporting, so a concurrent
# request may have filled the slot in the meantime.
if len(self._models) >= self._max_models:
raise ModelCapacityError(f"Maximum models per session reached ({self._max_models})")
self._models[model_id] = model
self._raws[model_id] = merged_raw
self._graphs[model_id] = artifact
self._summaries[model_id] = summary
if content_hash is not None:
# If a concurrent request beat us to it, the last writer wins;
# the race is benign (both models work, the older one is just
# not reachable via the index). See PLAN_model_load_dedup.md §6.3.
self._content_hash_index[content_hash] = model_id
return LoadResult(
model_id=model_id,
data_objects=summary.data_objects,
dimensions=summary.dimensions,
measures=summary.measures,
metrics=summary.metrics,
warnings=warnings,
model_load="fresh",
health=self._health_for(model),
)
def get_model(self, model_id: str) -> SemanticModel:
"""Look up a loaded model. Raises ``KeyError`` if not found."""
with self._lock:
try:
return self._models[model_id]
except KeyError:
raise KeyError(f"No model loaded with id '{model_id}'") from None
def get_raw(self, model_id: str) -> dict[str, object]:
"""Return the raw OBML dict for a loaded model.
Prefers the merged raw dict captured verbatim at load time (so
every field round-trips intact). Falls back to the lossy
``_model_to_raw`` reconstruction only for models that never passed
through :meth:`load_model` (e.g. programmatically constructed).
Raises ``KeyError`` if no model is loaded with the given id.
Returns a deep copy so callers (and downstream converters) cannot
mutate the store's internal ``_raws`` entry, which would corrupt
later exports or ``inherits`` merges.
"""
with self._lock:
if model_id not in self._models:
raise KeyError(f"No model loaded with id '{model_id}'")
raw = self._raws.get(model_id)
model = self._models[model_id]
if raw is None:
return self._model_to_raw(model)
return copy.deepcopy(raw)
def describe(self, model_id: str) -> ModelDescription:
"""Return a structured summary suitable for LLM consumption."""
model = self.get_model(model_id)
data_objects = [
DataObjectInfo(
label=obj.name,
code=obj.qualified_code,
columns=list(obj.columns.keys()),
join_targets=[j.join_to for j in obj.joins],
synonyms=obj.synonyms,
owner=obj.owner,
data_source=obj.data_source,
)
for obj in model.data_objects.values()
]
dimensions = [
DimensionInfo(
name=dim.name,
result_type=dim.result_type.value,
data_object=dim.view,
column=dim.column,
time_grain=dim.time_grain.value if dim.time_grain else None,
synonyms=dim.synonyms,
owner=dim.owner,
)
for dim in model.dimensions.values()
]
measures = [
MeasureInfo(
name=m.name,
result_type=m.result_type.value,
aggregation=m.aggregation,
expression=m.expression,
synonyms=m.synonyms,
owner=m.owner,
)
for m in model.measures.values()
]
metrics = [
MetricInfo(
name=met.name,
expression=met.expression,
synonyms=met.synonyms,
type=met.type.value,
measure=met.measure,
time_dimension=met.time_dimension,
owner=met.owner,
)
for met in model.metrics.values()
]
return ModelDescription(
model_id=model_id,
data_objects=data_objects,
dimensions=dimensions,
measures=measures,
metrics=metrics,
)
def list_models(self) -> list[ModelSummary]:
"""Return a short summary for every loaded model."""
with self._lock:
return list(self._summaries.values())
def remove_model(self, model_id: str) -> None:
"""Unload a model and its cached OBSL graph. Raises ``KeyError`` if not found.
Also removes the model's entry from the dedup index so the next load
of the same OBML content runs fresh. PLAN_model_load_dedup.md §6.2.
"""
with self._lock:
try:
del self._models[model_id]
except KeyError:
raise KeyError(f"No model loaded with id '{model_id}'") from None
self._raws.pop(model_id, None)
self._graphs.pop(model_id, None)
self._summaries.pop(model_id, None)
stale_hashes = [h for h, mid in self._content_hash_index.items() if mid == model_id]
for h in stale_hashes:
del self._content_hash_index[h]
was_shared = model_id in self._shared_ids
self._shared_ids.discard(model_id)
# Release the shared reference outside the store lock. The shared model
# stays compiled while any other session still references it.
if was_shared and self._shared_cache is not None:
self._shared_cache.release(model_id)
def close(self) -> None:
"""Drop this store, releasing every shared reference it holds.
Called by :class:`SessionManager` when a session is purged/expired/
closed, so shared models are refcount-released and can be evicted once
no live session references them. Private (non-shared) models are simply
discarded with the store.
"""
with self._lock:
shared_ids = list(self._shared_ids)
self._shared_ids.clear()
self._models.clear()
self._raws.clear()
self._graphs.clear()
self._summaries.clear()
self._content_hash_index.clear()
if self._shared_cache is not None:
for mid in shared_ids:
self._shared_cache.release(mid)
def compile_query(
self,
model_id: str,
query: QueryObject,
dialect: str,
) -> CompilationResult:
"""Compile a query against a loaded model."""
model = self.get_model(model_id)
if self._compilation_cache is None:
return self._pipeline.compile(query, model, dialect)
return self._compilation_cache.compile(self._pipeline, query, model, dialect)
def refresh_contracts(self, model_id: str) -> dict[str, RefreshContract]:
"""Per-physical-table freshness contracts for the given model.
Used by the result cache to derive an effective TTL for a query
based on the dataObjects it touched.
"""
from orionbelt.cache.contracts import collect_table_contracts
model = self.get_model(model_id)
contracts, _ = collect_table_contracts(model)
return contracts
def validate(
self,
yaml_str: str | None = None,
*,
raw_dict: dict[str, object] | None = None,
extends_yaml: list[str] | None = None,
inherits_model_id: str | None = None,
datasource_dialect: str | None = None,
follow_model_source: bool = False,
) -> ValidationSummary:
"""Validate a model without storing it. Accepts YAML string or raw dict.
Reports JSON Schema violations alongside semantic errors so that the
``/validate`` endpoints match what the schema-guarded load/query
endpoints enforce — a model that fails the schema is reported invalid
here rather than being silently coerced.
``datasource_dialect`` turns on the online check: every data object is
probed against the configured warehouse for that dialect and any drift
— a dropped table, a renamed column, a column whose type no longer
matches ``abstractType`` — is reported as an error alongside the
offline ones. Left as ``None``, validation stays entirely offline and
opens no connection. With ``follow_model_source``, ``datasource_dialect``
is only a default: a model naming a configured data source is probed
with that source's dialect, the connection its queries open.
"""
_model, raw, errors, warnings = self._parse_and_validate(
yaml_str,
raw_dict=raw_dict,
extends_yaml=extends_yaml,
inherits_model_id=inherits_model_id,
)
# Schema-validate the already-safely-parsed dict (``_parse_and_validate``
# loads via the safety-checked TrackedLoader). Skip when the YAML never
# parsed safely — the fatal parse/safety error is already reported and
# there is no document to validate.
fatal = {"YAML_SAFETY_ERROR", "YAML_PARSE_ERROR"}
if not any(e.code in fatal for e in errors):
errors = self._schema_errors(raw) + errors
if datasource_dialect:
if follow_model_source:
from orionbelt.service.datasource_selection import named_source_dialect
datasource_dialect = named_source_dialect(_model) or datasource_dialect
errors = errors + self._datasource_errors(_model, datasource_dialect)
return ValidationSummary(
valid=len(errors) == 0,
errors=errors,
warnings=warnings,
)
@staticmethod
def _datasource_errors(model: SemanticModel, dialect: str) -> list[ErrorInfo]:
"""Online findings for *model*, as ``ErrorInfo``.
Imported here rather than at module scope so that the offline path —
which is every caller that does not ask for the check — never pulls in
the executor and its driver stack.
"""
from orionbelt.service.datasource_probe import probe_datasource
return [
ErrorInfo(
code=f.code,
message=f.message,
path=f.path,
severity=f.severity,
hint=f.hint,
context=f.context,
)
for f in probe_datasource(model, dialect=dialect)
]
@staticmethod
def _schema_errors(raw: dict[str, object]) -> list[ErrorInfo]:
"""JSON Schema errors for a parsed model document, as ``ErrorInfo``."""
public = {k: v for k, v in raw.items() if not str(k).startswith("_")}
return [
ErrorInfo(code=e.code, message=e.message, path=e.path, severity=e.severity)
for e in validate_obml_document(public)
]
# -- OBSL graph ---------------------------------------------------------
def get_graph(self, model_id: str) -> GraphArtifact:
"""Return the cached OBSL graph for a model. Raises ``KeyError`` if not found."""
with self._lock:
try:
return self._graphs[model_id]
except KeyError:
raise KeyError(f"No graph for model '{model_id}'") from None
def query_graph(self, model_id: str, sparql: str) -> SPARQLResult:
"""Execute a read-only SPARQL query against a model's OBSL graph."""
artifact = self.get_graph(model_id)
return execute_sparql(artifact.graph, sparql)