"""marivo.semantic - Python-native semantic layer (v1.1).
Public surface::
import marivo.datasource as md
import marivo.semantic as ms
catalog = ms.load() # returns SemanticCatalog
catalog = ms.load(domains=['sales']) # filter to specific domains
catalog.domains.show()
catalog.metrics.show() # all metrics across domains
ms.domain(name="sales", owner="Mina Zhang", default=True)
warehouse = md.duckdb("warehouse").ref
orders = ms.entity(name="orders", datasource=warehouse, source=md.table("orders"))
amount = ms.measure_column(
name="amount", entity=orders, column="amount",
additivity="additive", unit="USD",
)
revenue = ms.aggregate(name="revenue", measure=amount, agg="sum")
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from marivo.refs import (
PeriodCalendarKind,
Ref,
SemanticKind,
TemporalSetKind,
WorkScheduleKind,
ref,
)
from marivo.semantic import errors as errors
from marivo.semantic import typing as typing
from marivo.semantic.authoring import (
GrainToDate,
PeriodCorrespondence,
aggregate,
ai_context,
all_rows,
bind,
calendar_grain,
count,
cumulative,
datetime,
dimension,
dimension_column,
domain,
entity,
event,
from_sql,
grain_to_date,
hour_prefix,
inception,
join_on,
lifecycle_state,
linear,
measure,
measure_column,
metric,
model_state,
participant,
participant_role,
period_calendar,
period_correspondence,
ratio,
relationship,
semi_additive,
snapshot,
state_model,
strptime,
temporal_set,
time_dimension,
time_dimension_column,
timestamp,
trailing,
transition,
validity,
weighted_mean,
where,
work_schedule,
)
from marivo.semantic.catalog import (
CalendarLevelDetails,
CalendarPeriodPage,
CatalogCollection,
CatalogEntry,
DatasourceDetails,
DatasourceEntry,
DerivedMetricDetails,
DimensionDetails,
DimensionEntry,
DomainDetails,
DomainEntry,
EntityDetails,
EntityEntry,
EventDetails,
EventEntry,
MeasureDetails,
MeasureEntry,
MetricDetails,
MetricEntry,
PeriodCalendarDetails,
PeriodCalendarEntry,
RelationshipDetails,
RelationshipEntry,
SemanticCatalog,
SimpleMetricDetails,
StateModelDetails,
StateModelEntry,
TemporalOccurrencePage,
TemporalSetDetails,
TemporalSetEntry,
TimeDimensionDetails,
TimeDimensionEntry,
WorkScheduleDetails,
WorkScheduleEntry,
load,
)
from marivo.semantic.dtos import PreviewBatchResult, VerifyResult
from marivo.semantic.event import Participant, ParticipantRoleHandle
from marivo.semantic.ir import (
AggregateFoldInput,
AggregateFoldValue,
JoinKey,
SqlProvenance,
)
from marivo.semantic.parity import ParityResult
from marivo.semantic.readiness import (
ReadinessInputSummary,
ReadinessIssue,
ReadinessReport,
)
from marivo.semantic.richness import RichnessReport
from marivo.semantic.state_model import (
Inception,
LifecycleState,
ModelStateHandle,
StateTransition,
)
from marivo.semantic.typing import AiContextValue
if TYPE_CHECKING:
from marivo.semantic.richness import DemandSignal
[docs]
def richness(
*,
demand: DemandSignal | None = None,
) -> RichnessReport:
"""Return a demand-ranked advisory richness report.
Pure advisory: it never blocks and never mutates readiness. ``demand``
seeds coverage/depth ranking from example questions, analysis intents,
run-history refs, and the build purpose.
Args:
demand: Optional demand signal for ranking richness gaps.
Returns:
RichnessReport with demand-ranked coverage and depth gaps.
Example:
>>> import marivo.semantic as ms
>>> report = ms.richness()
>>> report.show()
Constraints:
Advisory only — does not block readiness certification or runtime analysis.
"""
from marivo.semantic.reader import SemanticProject
project = SemanticProject()
project.load()
return project.richness(demand=demand)
[docs]
def parity_check(
name: str,
*,
rel_tol: float | None = None,
abs_tol: float | None = None,
force: bool = False,
) -> ParityResult:
"""Run parity check for a metric against its source SQL.
Datasource backends are resolved internally via the connection service.
Args:
name: Fully qualified metric ref (e.g. ``"sales.revenue"``).
rel_tol: Relative tolerance for numeric comparison. None uses default.
abs_tol: Absolute tolerance for numeric comparison. None uses default.
force: If True, re-runs parity even if cached results exist.
Returns:
ParityResult with comparison details and pass/fail status.
Example:
>>> import marivo.semantic as ms
>>> result = ms.parity_check("sales.revenue")
>>> result.show()
Constraints:
Requires the metric to declare ``provenance=ms.from_sql(sql=..., dialect=...)``.
Raises ``SemanticRuntimeError`` if the metric has no provenance.
"""
from marivo.semantic.reader import SemanticProject
project = SemanticProject()
project.load()
return project.parity_check(name, rel_tol=rel_tol, abs_tol=abs_tol, force=force)
__all__ = [
"AggregateFoldInput",
"AggregateFoldValue",
"AiContextValue",
"CalendarLevelDetails",
"CalendarPeriodPage",
"CatalogCollection",
"CatalogEntry",
"DatasourceDetails",
"DatasourceEntry",
"DerivedMetricDetails",
"DimensionDetails",
"DimensionEntry",
"DomainDetails",
"DomainEntry",
"EntityDetails",
"EntityEntry",
"EventDetails",
"EventEntry",
"GrainToDate",
"Inception",
"JoinKey",
"LifecycleState",
"MeasureDetails",
"MeasureEntry",
"MetricDetails",
"MetricEntry",
"ModelStateHandle",
"ParityResult",
"Participant",
"ParticipantRoleHandle",
"PeriodCalendarDetails",
"PeriodCalendarEntry",
"PeriodCalendarKind",
"PeriodCorrespondence",
"PreviewBatchResult",
"ReadinessInputSummary",
"ReadinessIssue",
"ReadinessReport",
"Ref",
"RelationshipDetails",
"RelationshipEntry",
"RichnessReport",
"SemanticCatalog",
"SemanticKind",
"SimpleMetricDetails",
"SqlProvenance",
"StateModelDetails",
"StateModelEntry",
"StateTransition",
"TemporalOccurrencePage",
"TemporalSetDetails",
"TemporalSetEntry",
"TemporalSetKind",
"TimeDimensionDetails",
"TimeDimensionEntry",
"VerifyResult",
"WorkScheduleDetails",
"WorkScheduleEntry",
"WorkScheduleKind",
"aggregate",
"ai_context",
"all_rows",
"bind",
"calendar_grain",
"count",
"cumulative",
"datetime",
"dimension",
"dimension_column",
"domain",
"entity",
"errors",
"event",
"from_sql",
"grain_to_date",
"hour_prefix",
"inception",
"join_on",
"lifecycle_state",
"linear",
"load",
"measure",
"measure_column",
"metric",
"model_state",
"parity_check",
"participant",
"participant_role",
"period_calendar",
"period_correspondence",
"ratio",
"ref",
"relationship",
"richness",
"semi_additive",
"snapshot",
"state_model",
"strptime",
"temporal_set",
"time_dimension",
"time_dimension_column",
"timestamp",
"trailing",
"transition",
"typing",
"validity",
"weighted_mean",
"where",
"work_schedule",
]
def _install_telemetry() -> None:
import sys
from marivo.semantic._capabilities.registry import REGISTRY
from marivo.telemetry import install_surface_instrumentation
install_surface_instrumentation(
surface="semantic",
descriptors=REGISTRY._descriptors,
root_module=sys.modules[__name__],
)
_install_telemetry()