diff --git a/README.md b/README.md index 5a18401..cdc253a 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ SkyData Studio is the post-ingestion data engineering application in the Sky ecosystem. It begins where SkyCommand's ingestion responsibility ends and prepares trusted data for analytical consumption by SkyWeb Analytics, Power BI, and future client-facing applications. -**Current status:** Phases 0 through 7 and Phase 8.1 are complete. Phase 8.1 closed with a READY 11-node/10-edge federated graph, a DFF impact radius reaching all 10 downstream nodes, and final validation passing 76 tests. **Phase 8.2 is now in progress:** source-to-target field mappings, dbt column annotations, and governed metric expressions are stitched into a field-level impact graph without creating a second lineage authority. +**Current status:** Phases 0 through 8 are complete. Phase 8 closed with source-controlled analytics-consumer lineage resolving one declared report against all four governed Federal Funds Rate metrics, with a 5-node/4-edge consumer graph and metric-level downstream impact. **Phase 9.1 is now in progress:** analytical mart publication readiness composes physical relation evidence, dbt build state, source-to-mart freshness, quality policy, semantic resolution, and declared consumers into one explicit delivery gate. --- @@ -406,8 +406,8 @@ The public roadmap is intentionally compact so GitHub visitors can see progress | Phase 5 | βœ… Complete | Apache Airflow 3 durable batch orchestration, REST API v2 integration, DAG/run/task observability, schedules, backfills, and ingestion-complete triggers | | Phase 6 | βœ… Complete | dbt transformation and modelling foundation with tested staging, intermediate, mart, and semantic layers | | Phase 7 | βœ… Complete | Data quality, reconciliation, blocking policies, incidents, SLOs, and observability | -| Phase 8 | πŸ”„ In Progress | Asset/field/model/report lineage and downstream impact analysis | -| Phase 9 | ⏳ Planned | Curated analytical marts, governed metrics, semantic delivery, and SkyWeb consumer contracts | +| Phase 8 | βœ… Complete | Asset/field/model/report lineage, trust/runtime overlays, and downstream impact analysis | +| Phase 9 | πŸ”„ In Progress | Curated analytical marts, publication readiness, governed semantic delivery, and downstream consumer products | | Phase 10 | ⏳ Planned | Power BI semantic models, refresh strategy, reporting inventory, and governed executive delivery | | Phase 11 | ⏳ Planned | Security, RBAC, environment promotion, audit evidence, backup, recovery, and retention | | Phase 12 | ⏳ Planned | AI-assisted mapping, SQL, tests, documentation, and explain-before-apply engineering workflows | @@ -520,11 +520,11 @@ The repository currently includes: - a completed Phase 8.2 field-level lineage graph from DFF source fields through dbt derivations to governed metrics; - a completed Phase 8.3 trust overlay that projects dbt checks, quality-contract rules, and durable incidents onto exact lineage nodes; - a completed Phase 8.4 runtime-lineage workbench that links the latest Airflow DAG/task execution to its replay-safe Studio pipeline run, structured steps, source asset, and curated target; -- a Phase 8.5 analytics-consumer lineage surface that resolves source-controlled report dependencies against governed dbt semantic metrics without claiming downstream deployment; +- a completed Phase 8.5 analytics-consumer lineage surface that resolves source-controlled report dependencies against governed dbt semantic metrics without claiming downstream deployment; +- an active Phase 9.1 analytical-product gate that compares the curated source with its dbt mart and composes freshness, quality, semantic, and consumer evidence before publication; - backend/frontend validation, repository map, and compact handoff tooling. -The active implementation target is **Phase 8.5 β€” Analytics Consumer Lineage and Impact Closure**. Phases 0 through 7 are complete; Phase 8.1 established asset impact, Phase 8.2 added field impact, Phase 8.3 projected trust evidence, and Phase 8.4 joined Airflow/Studio runtime execution to the structural graph. Phase 8.5 now closes the lineage chain at source-controlled analytics consumers while keeping actual analytical delivery and Power BI deployment in their later roadmap phases. - +The active implementation target is **Phase 9.1 β€” Analytical Mart Publication Readiness and Freshness Gate**. Phases 0 through 8 are complete. Phase 9.1 converts the proven Federal Funds Rate mart into a governed analytical product whose publication state becomes READY only when physical source/mart evidence, dbt build status, freshness alignment, quality policy, semantic definitions, and declared consumers all resolve together. ## Phase 8.5 quick start @@ -537,3 +537,29 @@ Invoke-RestMethod ` ``` A complete proof reports one resolved consumer, four resolved metrics, zero unresolved dependencies, 5 nodes, and 4 `CONSUMED_BY` edges. + +Phase 8.5 closure evidence: + +- `GET /api/v1/lineage/consumers/summary` returned `READY` against `READY` semantic artifacts with 1/1 consumer contracts resolved, 4/4 governed metrics resolved, zero unresolved metrics, and a 5-node/4-edge consumer graph; +- `average_federal_funds_rate` reported exactly one downstream consumer, `Federal Funds Rate Overview`, targeting Power BI with deployment status `DECLARED`; +- the Lineage workbench rendered consumers 1/1, metrics 4/4, semantic READY, and the explicit declaration-only delivery boundary; +- local code/test/build validation passed Ruff, mypy across 60 source files, 92 pytest tests, ESLint, and Vite production build; +- `npm ci` also surfaced one high-severity dependency advisory. The feature phase is functionally closed, but dependency-security maintenance remains an explicit follow-up rather than being misreported as a clean audit. + +## Phase 9.1 quick start + +Phase 9.1 introduces the first source-controlled analytical-product contract at `contracts/analytics/products/fed_funds_rate_daily_product.v1.json`. Studio reads the curated source relation, the dbt mart, generated dbt artifacts, the Phase 7 quality contract, and the Phase 8.5 consumer declaration at request time. It does not execute dbt automatically. + +```powershell +Invoke-RestMethod ` + "http://localhost:8100/api/v1/analytics/products/summary" | + ConvertTo-Json -Depth 12 +``` + +If ingestion has advanced `mart.fed_funds_rate` since the latest dbt build, the correct first result is `product_status=STALE` with the source row/date evidence ahead of `dbt_mart.fct_fed_funds_rate_daily`. Refresh the transformation explicitly: + +```powershell +.\scripts\dbt.ps1 build +``` + +Then refresh the endpoint or **Analytics Delivery β†’ Analytical Marts**. Publication becomes `READY` only after all six gates pass: physical relations, dbt build evidence, freshness alignment, quality contract, semantic product resolution, and declared-consumer resolution. diff --git a/apps/api/skydata_studio/api/router.py b/apps/api/skydata_studio/api/router.py index 5296f5a..d5c5e48 100644 --- a/apps/api/skydata_studio/api/router.py +++ b/apps/api/skydata_studio/api/router.py @@ -2,6 +2,7 @@ from skydata_studio.api.routes import ( airflow, + analytics, contracts, dbt, health, @@ -16,6 +17,7 @@ api_router = APIRouter() api_router.include_router(health.router, prefix="/health", tags=["health"]) +api_router.include_router(analytics.router, prefix="/analytics", tags=["analytics-delivery"]) api_router.include_router( airflow.router, prefix="/integrations/airflow", diff --git a/apps/api/skydata_studio/api/routes/analytics.py b/apps/api/skydata_studio/api/routes/analytics.py new file mode 100644 index 0000000..fc27a5c --- /dev/null +++ b/apps/api/skydata_studio/api/routes/analytics.py @@ -0,0 +1,21 @@ +from fastapi import APIRouter, HTTPException, status + +from skydata_studio.db.session import SessionDependency +from skydata_studio.schemas.analytics import AnalyticsProductSummary +from skydata_studio.services.analytics_products import analytics_product_summary + +router = APIRouter() + + +@router.get("/products/summary", response_model=AnalyticsProductSummary) +def product_summary(session: SessionDependency) -> AnalyticsProductSummary: + try: + return analytics_product_summary(session) + except (OSError, ValueError, TypeError) as error: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=( + "SkyData Studio could not compose analytical-product readiness. " + "Verify product contracts, dbt artifacts, and Studio PostgreSQL." + ), + ) from error diff --git a/apps/api/skydata_studio/api/routes/platform.py b/apps/api/skydata_studio/api/routes/platform.py index 8812317..d62100a 100644 --- a/apps/api/skydata_studio/api/routes/platform.py +++ b/apps/api/skydata_studio/api/routes/platform.py @@ -63,14 +63,14 @@ "Stitches mappings, dbt and field dependencies, semantic delivery, " "quality controls, incidents, and runtime execution into one impact graph." ), - status="FOUNDATION", + status="READY", phase=8, ), Capability( code="ANALYTICS_DELIVERY", name="Analytics Delivery", description="Publishes curated products to SkyWeb Analytics and Power BI.", - status="PLANNED", + status="FOUNDATION", phase=9, ), ] @@ -84,8 +84,8 @@ RoadmapPhase(number=5, name="Apache Airflow Integration", status="COMPLETE"), RoadmapPhase(number=6, name="dbt Transformation and Modelling", status="COMPLETE"), RoadmapPhase(number=7, name="Data Quality and Observability", status="COMPLETE"), - RoadmapPhase(number=8, name="Lineage and Impact Analysis", status="IN_PROGRESS"), - RoadmapPhase(number=9, name="Analytical Marts and Semantic Delivery", status="PLANNED"), + RoadmapPhase(number=8, name="Lineage and Impact Analysis", status="COMPLETE"), + RoadmapPhase(number=9, name="Analytical Marts and Semantic Delivery", status="IN_PROGRESS"), RoadmapPhase(number=10, name="Power BI Integration", status="PLANNED"), ] @@ -96,7 +96,9 @@ def platform_summary() -> PlatformSummary: product="SkyData Studio", subtitle="Data Engineering Workbench", theme="Aurora Foundry", - current_phase="Phase 8.5 β€” Analytics Consumer Lineage and Impact Closure", + current_phase=( + "Phase 9.1 β€” Analytical Mart Publication Readiness and Freshness Gate" + ), boundary=( "SkyData Studio starts after SkyCommand ingestion and publishes governed " "analytical products for SkyWeb Analytics, Power BI, and future consumers." diff --git a/apps/api/skydata_studio/schemas/analytics.py b/apps/api/skydata_studio/schemas/analytics.py new file mode 100644 index 0000000..d52bad5 --- /dev/null +++ b/apps/api/skydata_studio/schemas/analytics.py @@ -0,0 +1,68 @@ +from typing import Literal + +from pydantic import BaseModel, Field + +AnalyticsProductStatus = Literal["READY", "STALE", "BLOCKED", "PENDING", "MISSING"] +AnalyticsGateStatus = Literal["PASS", "WARN", "BLOCK", "PENDING"] +AnalyticsFreshnessStatus = Literal["ALIGNED", "STALE", "MISSING", "UNKNOWN"] + + +class AnalyticsProductDefinition(BaseModel): + code: str + version: str + name: str + description: str + owner: str + domain: str + source_relation: str + mart_relation: str + freshness_column: str + row_alignment: Literal["EXACT"] = "EXACT" + semantic_model: str + quality_contract_code: str + required_metrics: list[str] + required_dimensions: list[str] + consumer_codes: list[str] + publication_mode: Literal["GATED"] = "GATED" + + +class AnalyticsRelationEvidence(BaseModel): + relation: str + status: Literal["READY", "MISSING"] + row_count: int | None = Field(default=None, ge=0) + max_freshness_value: str | None = None + + +class AnalyticsPublicationGate(BaseModel): + code: str + label: str + status: AnalyticsGateStatus + message: str + + +class AnalyticsProductSummary(BaseModel): + project_name: str = "skydata_studio" + phase: str = "Phase 9.1 β€” Analytical Mart Publication Readiness and Freshness Gate" + product_status: AnalyticsProductStatus + product_code: str + product_version: str + product_name: str + description: str + owner: str + domain: str + source_path: str + source_relation: AnalyticsRelationEvidence + mart_relation: AnalyticsRelationEvidence + row_count_delta: int | None = None + freshness_status: AnalyticsFreshnessStatus + refresh_required: bool + model_build_status: Literal["READY", "ERROR", "UNKNOWN", "MISSING"] + semantic_artifact_status: Literal["READY", "PENDING", "MISSING"] + semantic_model_resolved: bool + quality_contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"] + required_metric_count: int = Field(ge=0) + resolved_metric_count: int = Field(ge=0) + required_consumer_count: int = Field(ge=0) + resolved_consumer_count: int = Field(ge=0) + gates: list[AnalyticsPublicationGate] + publication_message: str diff --git a/apps/api/skydata_studio/services/analytics_products.py b/apps/api/skydata_studio/services/analytics_products.py new file mode 100644 index 0000000..ab6aac1 --- /dev/null +++ b/apps/api/skydata_studio/services/analytics_products.py @@ -0,0 +1,429 @@ +from __future__ import annotations + +import json +import re +from datetime import date, datetime +from pathlib import Path +from typing import Any, Literal, cast + +from sqlalchemy import inspect, text +from sqlalchemy.orm import Session + +from skydata_studio.schemas.analytics import ( + AnalyticsProductDefinition, + AnalyticsProductSummary, + AnalyticsPublicationGate, + AnalyticsRelationEvidence, +) +from skydata_studio.schemas.dbt import ( + DbtModelCatalogueSummary, + DbtSemanticLayerSummary, +) +from skydata_studio.schemas.lineage import AnalyticsConsumerLineageSummary +from skydata_studio.schemas.quality import QualityContractSummary +from skydata_studio.services.dbt_transformations import ( + dbt_model_catalogue, + dbt_semantic_layer, +) +from skydata_studio.services.lineage_consumers import analytics_consumer_lineage_summary +from skydata_studio.services.quality_contracts import quality_contract_summary + +_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def _repository_root() -> Path: + return Path(__file__).resolve().parents[4] + + +def _default_product_path() -> Path: + return ( + _repository_root() + / "contracts" + / "analytics" + / "products" + / "fed_funds_rate_daily_product.v1.json" + ) + + +def _consumer_contract_directory() -> Path: + return _repository_root() / "contracts" / "analytics" + + +def _read_product(path: Path) -> AnalyticsProductDefinition: + with path.open(encoding="utf-8") as handle: + payload = json.load(handle) + if not isinstance(payload, dict): + raise ValueError(f"Expected an object in analytical product contract {path}.") + return AnalyticsProductDefinition.model_validate(cast(dict[str, Any], payload)) + + +def _relation_parts(relation: str) -> tuple[str, str]: + parts = relation.split(".") + if len(parts) != 2 or not all(_IDENTIFIER.fullmatch(part) for part in parts): + raise ValueError(f"Unsafe analytical relation identifier: {relation!r}.") + return parts[0], parts[1] + + +def _column_name(column: str) -> str: + if not _IDENTIFIER.fullmatch(column): + raise ValueError(f"Unsafe analytical freshness column: {column!r}.") + return column + + +def _scalar_to_text(value: object) -> str | None: + if value is None: + return None + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, date): + return value.isoformat() + return str(value) + + +def relation_evidence( + session: Session, + relation: str, + freshness_column: str, +) -> AnalyticsRelationEvidence: + schema, relation_name = _relation_parts(relation) + column = _column_name(freshness_column) + inspector = inspect(session.get_bind()) + if not inspector.has_table(relation_name, schema=schema): + return AnalyticsRelationEvidence(relation=relation, status="MISSING") + + count_statement = text(f'SELECT COUNT(*) FROM "{schema}"."{relation_name}"') + freshness_statement = text( + f'SELECT MAX("{column}") FROM "{schema}"."{relation_name}"' + ) + row_count = int(session.execute(count_statement).scalar_one()) + freshness_value = session.execute(freshness_statement).scalar_one() + return AnalyticsRelationEvidence( + relation=relation, + status="READY", + row_count=row_count, + max_freshness_value=_scalar_to_text(freshness_value), + ) + + +def _model_build_status( + product: AnalyticsProductDefinition, + catalogue: DbtModelCatalogueSummary, +) -> Literal["READY", "ERROR", "UNKNOWN", "MISSING"]: + model = next( + (item for item in catalogue.models if item.relation == product.mart_relation), + None, + ) + if model is None: + return "MISSING" + return model.build_status + + +def _semantic_resolution( + product: AnalyticsProductDefinition, + semantic: DbtSemanticLayerSummary, +) -> tuple[bool, int]: + semantic_model = next( + (item for item in semantic.semantic_models if item.name == product.semantic_model), + None, + ) + if semantic_model is None or semantic.artifact_status != "READY": + return False, 0 + + available_dimensions = {item.name for item in semantic_model.dimensions} + dimensions_ready = all( + name in available_dimensions for name in product.required_dimensions + ) + available_metrics = { + item.name + for item in semantic.metrics + if item.semantic_model == product.semantic_model + } + resolved_metric_count = sum( + metric_name in available_metrics for metric_name in product.required_metrics + ) + return ( + dimensions_ready and resolved_metric_count == len(product.required_metrics), + resolved_metric_count, + ) + + +def _consumer_resolution( + product: AnalyticsProductDefinition, + consumers: AnalyticsConsumerLineageSummary, +) -> tuple[int, int]: + required_codes = set(product.consumer_codes) + resolved_codes = { + str(node.metadata.get("consumer_code") or "") + for node in consumers.nodes + if node.node_type == "ANALYTICS_CONSUMER" and node.status == "READY" + } + return len(required_codes), len(required_codes & resolved_codes) + + +def _freshness_status( + product: AnalyticsProductDefinition, + source: AnalyticsRelationEvidence, + mart: AnalyticsRelationEvidence, +) -> Literal["ALIGNED", "STALE", "MISSING", "UNKNOWN"]: + if source.status == "MISSING" or mart.status == "MISSING": + return "MISSING" + if source.row_count is None or mart.row_count is None: + return "UNKNOWN" + if source.max_freshness_value is None or mart.max_freshness_value is None: + return "UNKNOWN" + + rows_aligned = source.row_count == mart.row_count + dates_aligned = source.max_freshness_value == mart.max_freshness_value + if product.row_alignment == "EXACT" and rows_aligned and dates_aligned: + return "ALIGNED" + return "STALE" + + +def _gate( + code: str, + label: str, + status: Literal["PASS", "WARN", "BLOCK", "PENDING"], + message: str, +) -> AnalyticsPublicationGate: + return AnalyticsPublicationGate( + code=code, + label=label, + status=status, + message=message, + ) + + +def compose_analytics_product_summary( + *, + product: AnalyticsProductDefinition, + source_path: str, + source: AnalyticsRelationEvidence, + mart: AnalyticsRelationEvidence, + catalogue: DbtModelCatalogueSummary, + semantic: DbtSemanticLayerSummary, + quality: QualityContractSummary, + consumers: AnalyticsConsumerLineageSummary, +) -> AnalyticsProductSummary: + model_status = _model_build_status(product, catalogue) + semantic_resolved, resolved_metric_count = _semantic_resolution(product, semantic) + required_consumer_count, resolved_consumer_count = _consumer_resolution(product, consumers) + freshness_status = _freshness_status(product, source, mart) + + row_count_delta = None + if source.row_count is not None and mart.row_count is not None: + row_count_delta = source.row_count - mart.row_count + + gates: list[AnalyticsPublicationGate] = [] + physical_ready = source.status == "READY" and mart.status == "READY" + gates.append( + _gate( + "PHYSICAL_RELATIONS", + "Source and mart relations exist", + "PASS" if physical_ready else "BLOCK", + ( + "Curated source and dbt analytical mart are both materialized." + if physical_ready + else "The curated source or dbt analytical mart relation is missing." + ), + ) + ) + + model_gate_status: Literal["PASS", "WARN", "BLOCK", "PENDING"] + if model_status == "READY": + model_gate_status = "PASS" + elif model_status == "UNKNOWN": + model_gate_status = "PENDING" + else: + model_gate_status = "BLOCK" + gates.append( + _gate( + "DBT_MODEL_BUILD", + "dbt mart build evidence is green", + model_gate_status, + f"Latest dbt model status is {model_status}.", + ) + ) + + freshness_gate_status: Literal["PASS", "WARN", "BLOCK", "PENDING"] + if freshness_status == "ALIGNED": + freshness_gate_status = "PASS" + freshness_message = ( + "Curated source and analytical mart have matching row counts and latest dates." + ) + elif freshness_status == "STALE": + freshness_gate_status = "BLOCK" + freshness_message = ( + "Curated source is ahead of the analytical mart; run dbt build before publication." + ) + elif freshness_status == "MISSING": + freshness_gate_status = "BLOCK" + freshness_message = "Freshness cannot be proven because a required relation is missing." + else: + freshness_gate_status = "PENDING" + freshness_message = "Freshness evidence is incomplete." + gates.append( + _gate( + "FRESHNESS_ALIGNMENT", + "Curated source and dbt mart are synchronized", + freshness_gate_status, + freshness_message, + ) + ) + + quality_matches = quality.contract_code == product.quality_contract_code + quality_pass = quality_matches and quality.contract_status == "COMPLIANT" + quality_gate_status: Literal["PASS", "WARN", "BLOCK", "PENDING"] + if quality_pass: + quality_gate_status = "PASS" + elif quality.contract_status == "PENDING": + quality_gate_status = "PENDING" + else: + quality_gate_status = "BLOCK" + gates.append( + _gate( + "QUALITY_CONTRACT", + "Quality contract permits consumption", + quality_gate_status, + ( + "Required quality contract is COMPLIANT." + if quality_pass + else ( + "Required quality contract does not match the analytical product." + if not quality_matches + else f"Quality contract is {quality.contract_status}." + ) + ), + ) + ) + + semantic_gate_status: Literal["PASS", "WARN", "BLOCK", "PENDING"] + if semantic_resolved: + semantic_gate_status = "PASS" + elif semantic.artifact_status == "READY": + semantic_gate_status = "BLOCK" + else: + semantic_gate_status = "PENDING" + gates.append( + _gate( + "SEMANTIC_PRODUCT", + "Semantic model and governed metrics resolve", + semantic_gate_status, + ( + f"Resolved {resolved_metric_count}/" + f"{len(product.required_metrics)} required metrics." + ), + ) + ) + + consumer_ready = ( + required_consumer_count > 0 + and resolved_consumer_count == required_consumer_count + ) + consumer_gate_status: Literal["PASS", "WARN", "BLOCK", "PENDING"] + if consumer_ready: + consumer_gate_status = "PASS" + elif consumers.consumer_status == "MISSING": + consumer_gate_status = "BLOCK" + else: + consumer_gate_status = "PENDING" + gates.append( + _gate( + "DECLARED_CONSUMERS", + "Declared consumers resolve to governed semantics", + consumer_gate_status, + f"Resolved {resolved_consumer_count}/{required_consumer_count} required consumers.", + ) + ) + + blocking_codes = {item.code for item in gates if item.status == "BLOCK"} + pending = any(item.status == "PENDING" for item in gates) + if not physical_ready: + product_status: Literal["READY", "STALE", "BLOCKED", "PENDING", "MISSING"] = ( + "MISSING" + ) + elif "FRESHNESS_ALIGNMENT" in blocking_codes: + product_status = "STALE" + elif blocking_codes: + product_status = "BLOCKED" + elif pending: + product_status = "PENDING" + else: + product_status = "READY" + + if product_status == "READY": + publication_message = ( + "Analytical product is synchronized, governed, semantically resolved, " + "and ready for delivery." + ) + elif product_status == "STALE": + publication_message = ( + "Analytical product is structurally healthy but stale relative to the curated source. " + "Run dbt build, then refresh this workbench." + ) + elif product_status == "BLOCKED": + publication_message = "Analytical publication is blocked by one or more governance gates." + elif product_status == "MISSING": + publication_message = "Analytical publication is missing required physical data products." + else: + publication_message = "Analytical publication is waiting for complete runtime evidence." + + return AnalyticsProductSummary( + product_status=product_status, + product_code=product.code, + product_version=product.version, + product_name=product.name, + description=product.description, + owner=product.owner, + domain=product.domain, + source_path=source_path, + source_relation=source, + mart_relation=mart, + row_count_delta=row_count_delta, + freshness_status=freshness_status, + refresh_required=product_status == "STALE", + model_build_status=model_status, + semantic_artifact_status=semantic.artifact_status, + semantic_model_resolved=semantic_resolved, + quality_contract_status=quality.contract_status, + required_metric_count=len(product.required_metrics), + resolved_metric_count=resolved_metric_count, + required_consumer_count=required_consumer_count, + resolved_consumer_count=resolved_consumer_count, + gates=gates, + publication_message=publication_message, + ) + + +def analytics_product_summary( + session: Session, + *, + product_path: Path | None = None, + target_dir: Path | None = None, + consumer_contract_directory: Path | None = None, +) -> AnalyticsProductSummary: + path = product_path or _default_product_path() + product = _read_product(path) + try: + source_path = path.relative_to(_repository_root()).as_posix() + except ValueError: + source_path = path.as_posix() + + source = relation_evidence(session, product.source_relation, product.freshness_column) + mart = relation_evidence(session, product.mart_relation, product.freshness_column) + catalogue = dbt_model_catalogue(target_dir) + semantic = dbt_semantic_layer(target_dir) + quality = quality_contract_summary(target_dir=target_dir) + consumers = analytics_consumer_lineage_summary( + contract_directory=consumer_contract_directory or _consumer_contract_directory(), + target_dir=target_dir, + ) + return compose_analytics_product_summary( + product=product, + source_path=source_path, + source=source, + mart=mart, + catalogue=catalogue, + semantic=semantic, + quality=quality, + consumers=consumers, + ) diff --git a/apps/web/src/App.jsx b/apps/web/src/App.jsx index 41ed63c..84059e6 100644 --- a/apps/web/src/App.jsx +++ b/apps/web/src/App.jsx @@ -13,6 +13,7 @@ import SchedulesBackfills from './pages/SchedulesBackfills.jsx'; import Transformations from './pages/Transformations.jsx'; import DataModels from './pages/DataModels.jsx'; import SemanticLayer from './pages/SemanticLayer.jsx'; +import AnalyticalMarts from './pages/AnalyticalMarts.jsx'; import DataQuality from './pages/DataQuality.jsx'; import DataContracts from './pages/DataContracts.jsx'; import QualityIncidents from './pages/QualityIncidents.jsx'; @@ -75,6 +76,7 @@ function App() { } /> } /> } /> + } /> } /> } /> } /> @@ -96,6 +98,7 @@ function App() { '/orchestration/airflow', '/orchestration/runs', '/orchestration/backfills', + '/delivery/marts', '/delivery/semantic', '/quality/checks', '/quality/contracts', diff --git a/apps/web/src/components/Sidebar.jsx b/apps/web/src/components/Sidebar.jsx index f3f5f45..8afb446 100644 --- a/apps/web/src/components/Sidebar.jsx +++ b/apps/web/src/components/Sidebar.jsx @@ -122,7 +122,7 @@ function Sidebar({ mobileOpen, onClose }) { Development - Phase 8.5 consumer lineage + Phase 9.1 analytical mart readiness diff --git a/apps/web/src/pages/AnalyticalMarts.jsx b/apps/web/src/pages/AnalyticalMarts.jsx new file mode 100644 index 0000000..d545d04 --- /dev/null +++ b/apps/web/src/pages/AnalyticalMarts.jsx @@ -0,0 +1,241 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import StatusPill from '../components/StatusPill.jsx'; +import { getJson } from '../services/api.js'; + +const EMPTY = { + product_status: 'PENDING', + product_code: 'FED_FUNDS_RATE_DAILY_PRODUCT', + product_version: '1.0.0', + product_name: 'Federal Funds Rate Daily Analytical Product', + description: 'Waiting for analytical publication evidence.', + owner: 'SkyData Studio', + domain: 'Macroeconomics', + source_path: 'contracts/analytics/products/fed_funds_rate_daily_product.v1.json', + source_relation: { relation: 'mart.fed_funds_rate', status: 'MISSING', row_count: null, max_freshness_value: null }, + mart_relation: { relation: 'dbt_mart.fct_fed_funds_rate_daily', status: 'MISSING', row_count: null, max_freshness_value: null }, + row_count_delta: null, + freshness_status: 'UNKNOWN', + refresh_required: false, + model_build_status: 'UNKNOWN', + semantic_artifact_status: 'PENDING', + semantic_model_resolved: false, + quality_contract_status: 'PENDING', + required_metric_count: 4, + resolved_metric_count: 0, + required_consumer_count: 1, + resolved_consumer_count: 0, + gates: [], + publication_message: 'Waiting for analytical publication evidence.', +}; + +function formatCount(value) { + if (value === null || value === undefined) return 'β€”'; + return Number(value).toLocaleString(); +} + +function gateTone(status) { + if (status === 'PASS') return 'READY'; + if (status === 'WARN' || status === 'PENDING') return 'WARNING'; + if (status === 'BLOCK') return 'BLOCKED'; + return 'UNKNOWN'; +} + +function statusTone(status) { + if (status === 'READY' || status === 'ALIGNED' || status === 'COMPLIANT') return 'READY'; + if (status === 'STALE' || status === 'PENDING' || status === 'UNKNOWN') return 'WARNING'; + if (status === 'BLOCKED' || status === 'MISSING' || status === 'ERROR') return 'BLOCKED'; + return status; +} + +function AnalyticalMarts() { + const [summary, setSummary] = useState(EMPTY); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(''); + + const load = useCallback(async () => { + setLoading(true); + setError(''); + try { + setSummary(await getJson('/api/v1/analytics/products/summary')); + } catch (requestError) { + setError(requestError.message); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + const controller = new AbortController(); + getJson('/api/v1/analytics/products/summary', { signal: controller.signal }) + .then(setSummary) + .catch((requestError) => { + if (requestError.name !== 'AbortError') setError(requestError.message); + }) + .finally(() => { + if (!controller.signal.aborted) setLoading(false); + }); + return () => controller.abort(); + }, []); + + const passingGates = useMemo( + () => summary.gates.filter((gate) => gate.status === 'PASS').length, + [summary.gates], + ); + + return ( +
+
+
+ PHASE 9.1 Β· ANALYTICAL MART PUBLICATION READINESS +

Analytical Marts

+

+ Treat a mart as publishable only when physical data, dbt build evidence, freshness, + quality policy, semantic definitions, and declared consumers all agree. +

+
+ +
+ + {error ?
!{error}
: null} + +
+
+ Publication Gate + {summary.product_status} + {passingGates}/{summary.gates.length || 6} gates passing +
+
+ Source Rows + {formatCount(summary.source_relation.row_count)} + {summary.source_relation.relation} +
+
+ Mart Rows + {formatCount(summary.mart_relation.row_count)} + {summary.mart_relation.relation} +
+
+ Freshness + {summary.freshness_status} + {summary.row_count_delta === null ? 'Row delta pending' : `${summary.row_count_delta} row delta`} +
+
+ Metrics + {summary.resolved_metric_count}/{summary.required_metric_count} + Governed semantic measures +
+
+ Consumers + {summary.resolved_consumer_count}/{summary.required_consumer_count} + Declared delivery contracts +
+
+ +
+
+
+ ANALYTICAL PRODUCT +

{summary.product_name}

+
+ +
+

{summary.publication_message}

+ {summary.refresh_required ? ( +
+ Fresh dbt materialization required + Run .\scripts\dbt.ps1 build, then refresh this page. +
+ ) : null} +
+
PRODUCT{summary.product_code}v{summary.product_version}
+
DOMAIN{summary.domain}{summary.owner}
+
DBT BUILD{summary.model_build_status}Latest mart artifact state
+
QUALITY{summary.quality_contract_status}Source-controlled publication policy
+
SEMANTIC{summary.semantic_model_resolved ? 'RESOLVED' : 'PENDING'}{summary.semantic_artifact_status} artifacts
+
CONTRACTGATED{summary.source_path}
+
+
+ +
+
+
+ FRESHNESS EVIDENCE +

Curated source and analytical mart must move together

+
+ +
+
+
+ CURATED SOURCE +

{summary.source_relation.relation}

+ {formatCount(summary.source_relation.row_count)} rows + Latest {summary.source_relation.max_freshness_value || 'β€”'} +
+
+ β†’ + {summary.freshness_status} + {summary.row_count_delta === null ? 'delta pending' : `${summary.row_count_delta} rows`} +
+
+ DBT MART +

{summary.mart_relation.relation}

+ {formatCount(summary.mart_relation.row_count)} rows + Latest {summary.mart_relation.max_freshness_value || 'β€”'} +
+
+
+ +
+
+
+ PUBLICATION GATES +

Six independent proofs decide whether delivery is safe

+
+ {passingGates}/{summary.gates.length || 6} pass +
+
+ {summary.gates.map((gate, index) => ( +
+
+ 0{index + 1} Β· {gate.code.replaceAll('_', ' ')} + +
+

{gate.label}

+

{gate.message}

+
+ ))} +
+
+ +
+
+
+
ACCEPTANCE CONTRACT

What Phase 9.1 proves

+ 9.1 +
+
    +
  1. 01
    Freshness is a gateA healthy dbt model is not publishable when the curated source has moved ahead.
    2. +
  2. 02
    Governance composesPhysical, quality, semantic, and consumer evidence must all resolve together.
    3. +
  3. 03
    Publication stays explicitStudio reports readiness but does not silently execute dbt or deploy Power BI resources.
    4. +
+
+
+
+
BOUNDARY

Gate the product; do not hide the refresh

+ β–¦ +
+

+ Phase 9.1 turns the existing mart into a governed analytical product contract. dbt still + owns transformation execution, quality retains policy ownership, semantic definitions + remain dbt-owned, and downstream Power BI provisioning stays a later phase. +

+
Freshness proofGoverned deliveryNo hidden execution
+
+
+
+ ); +} + +export default AnalyticalMarts; diff --git a/apps/web/src/pages/StudioOverview.jsx b/apps/web/src/pages/StudioOverview.jsx index fb9bf20..2e45daf 100644 --- a/apps/web/src/pages/StudioOverview.jsx +++ b/apps/web/src/pages/StudioOverview.jsx @@ -5,7 +5,7 @@ const FALLBACK = { product: 'SkyData Studio', subtitle: 'Data Engineering Workbench', theme: 'Aurora Foundry', - current_phase: 'Phase 8.5 β€” Analytics Consumer Lineage and Impact Closure', + current_phase: 'Phase 9.1 β€” Analytical Mart Publication Readiness and Freshness Gate', boundary: 'SkyData Studio starts after SkyCommand ingestion and publishes governed analytical products for downstream consumers.', capabilities: [], }; @@ -41,7 +41,7 @@ function StudioOverview() {
- AURORA FOUNDRY Β· PHASE 8.5 + AURORA FOUNDRY Β· PHASE 9.1

Shape trusted data into analytical products.

{summary.boundary}

@@ -61,7 +61,7 @@ function StudioOverview() {
Platform API{apiState}FastAPI contract service
-
Current Phase8.5analytics consumer lineage closure
+
Current Phase9.1analytical mart publication readiness
Pipeline ProofREADYCurated mart materialization proven
OrchestratorAirflow 3Batch and asset workflows
@@ -96,8 +96,8 @@ function StudioOverview() { { code: 'AIRFLOW', name: 'Apache Airflow Orchestration', description: 'DAGs, assets, schedules, retries, and backfills.', status: 'FOUNDATION', phase: 5 }, { code: 'DBT', name: 'dbt Transformation Layer', description: 'Tested staging, intermediate, mart, and semantic models.', status: 'FOUNDATION', phase: 6 }, { code: 'QUALITY_LINEAGE', name: 'Data Quality and Reliability', description: 'Trust evidence, contracts, incidents, and reliability history.', status: 'READY', phase: 7 }, - { code: 'LINEAGE_IMPACT', name: 'Lineage and Impact Analysis', description: 'Federated asset and field lineage with trust and runtime execution context.', status: 'FOUNDATION', phase: 8 }, - { code: 'ANALYTICS_DELIVERY', name: 'Analytics Delivery', description: 'Governed products for SkyWeb Analytics and Power BI.', status: 'PLANNED', phase: 9 }, + { code: 'LINEAGE_IMPACT', name: 'Lineage and Impact Analysis', description: 'Federated asset, field, trust, runtime, and consumer impact analysis.', status: 'READY', phase: 8 }, + { code: 'ANALYTICS_DELIVERY', name: 'Analytics Delivery', description: 'Governed analytical marts and publication readiness for downstream consumers.', status: 'FOUNDATION', phase: 9 }, ]).map((capability) => (
Phase {capability.phase}
@@ -111,17 +111,17 @@ function StudioOverview() {
-
CURRENT IMPLEMENTATION

analytics consumer lineage and impact closure

PHASE 8.5
+
CURRENT IMPLEMENTATION

analytical mart publication readiness

PHASE 9.1
    -
  1. 01
    Airflow control-plane evidenceThe latest linked DAG run and task instances stay Airflow-owned and are read through REST API v2.
    2. -
  2. 02
    Replay-safe Studio executionThe AIRFLOW run key resolves to one durable pipeline run with four structured Studio step results.
    3. -
  3. 03
    Execution meets structural lineageThe runtime chain starts at DFF and terminates at the same curated mart node used by the design graph.
    4. +
  4. 01
    Freshness becomes explicitCurated-source and dbt-mart row/date evidence must align before publication is ready.
    5. +
  5. 02
    Governance composesQuality, semantic metrics, and declared consumers join the physical mart proof.
    6. +
  6. 03
    Delivery remains gatedStudio reports publication readiness without silently executing dbt or fabricating Power BI deployment.
ENGINEERING RULE

Contracts before coupling

∞
-

Studio consumes versioned APIs and approved read-only views. It does not write into SkyCommand schemas or read Airflow's internal metadata database.

-
Reusable architecturePortable domainsAuditable evidence
+

Studio composes evidence from physical relations, dbt artifacts, quality policy, semantic definitions, and consumer declarations while each authority remains independently owned.

+
Freshness proofGoverned deliveryAuditable evidence
diff --git a/apps/web/src/styles.css b/apps/web/src/styles.css index 5b91738..a84be9f 100644 --- a/apps/web/src/styles.css +++ b/apps/web/src/styles.css @@ -1945,3 +1945,96 @@ a { color: inherit; } grid-template-columns: 1fr; } } + +/* Phase 9.1 β€” analytical mart publication readiness */ +.analytical-product-metrics { grid-template-columns: repeat(6, minmax(0, 1fr)); } +.analytical-product-hero { position: relative; overflow: hidden; } +.analytical-product-hero::after { + content: ''; + position: absolute; + width: 260px; + height: 260px; + right: -130px; + top: -150px; + border-radius: 50%; + background: radial-gradient(circle, rgba(65, 236, 211, 0.12), transparent 68%); + pointer-events: none; +} +.analytical-product-message { margin: 0 0 18px; max-width: 920px; color: var(--muted); line-height: 1.65; } +.analytical-refresh-callout { + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; + margin-bottom: 18px; + padding: 13px 15px; + border: 1px solid rgba(242, 190, 75, 0.42); + border-radius: 12px; + background: rgba(242, 190, 75, 0.07); +} +.analytical-refresh-callout strong { color: #f5d27b; } +.analytical-refresh-callout span { color: var(--muted); } +.analytical-refresh-callout code { color: var(--text); } +.analytical-product-contract-grid { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + gap: 10px; +} +.analytical-product-contract-grid > div, +.analytical-relation-grid > article, +.analytical-gate-card { + border: 1px solid var(--border); + border-radius: 12px; + background: rgba(5, 20, 29, 0.62); +} +.analytical-product-contract-grid > div { padding: 13px 14px; display: grid; gap: 4px; } +.analytical-product-contract-grid span, +.analytical-relation-grid article > span, +.analytical-gate-card > div > span { + color: var(--muted-strong); + font-size: 0.67rem; + font-weight: 800; + letter-spacing: 0.1em; +} +.analytical-product-contract-grid strong { color: var(--aqua); overflow-wrap: anywhere; } +.analytical-product-contract-grid small { color: var(--muted); overflow-wrap: anywhere; } +.analytical-relation-grid { + display: grid; + grid-template-columns: minmax(0, 1fr) 150px minmax(0, 1fr); + gap: 14px; + align-items: stretch; +} +.analytical-relation-grid > article { padding: 18px; } +.analytical-relation-grid h3 { margin: 8px 0 16px; font-size: 1rem; color: var(--text); overflow-wrap: anywhere; } +.analytical-relation-grid article > strong { display: block; color: var(--aqua); font-size: 1.25rem; } +.analytical-relation-grid article > small { display: block; margin-top: 5px; color: var(--muted); } +.analytical-freshness-arrow { + display: grid; + place-content: center; + justify-items: center; + gap: 5px; + color: var(--muted); +} +.analytical-freshness-arrow > span { color: var(--aqua); font-size: 1.6rem; } +.analytical-freshness-arrow strong { color: var(--text); } +.analytical-freshness-arrow small { color: var(--muted); } +.analytical-gate-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 10px; } +.analytical-gate-card { padding: 15px; } +.analytical-gate-card > div { display: flex; align-items: center; justify-content: space-between; gap: 12px; } +.analytical-gate-card h3 { margin: 12px 0 6px; font-size: 0.95rem; } +.analytical-gate-card p { margin: 0; color: var(--muted); line-height: 1.5; } +.analytical-proof-grid { align-items: stretch; } + +@media (max-width: 1180px) { + .analytical-product-metrics { grid-template-columns: repeat(3, minmax(0, 1fr)); } +} + +@media (max-width: 760px) { + .analytical-product-metrics, + .analytical-product-contract-grid, + .analytical-gate-grid { grid-template-columns: 1fr; } + .analytical-relation-grid { grid-template-columns: 1fr; } + .analytical-freshness-arrow { min-height: 80px; } + .analytical-freshness-arrow > span { transform: rotate(90deg); } + .analytical-refresh-callout { align-items: flex-start; flex-direction: column; } +} diff --git a/contracts/analytics/README.md b/contracts/analytics/README.md index 4d8c6b3..2a19e10 100644 --- a/contracts/analytics/README.md +++ b/contracts/analytics/README.md @@ -1,7 +1,7 @@ -# Analytics consumer contracts +# Analytics contracts -This directory holds source-controlled declarations for downstream analytics consumers. +This directory holds source-controlled analytical-consumer and analytical-product declarations. -Phase 8.5 uses these declarations only for lineage and impact analysis. A declaration says which governed semantic model, metrics, and dimensions a consumer requires. It does **not** claim that a Power BI workspace, semantic model, report, refresh schedule, or deployment has already been provisioned. +Phase 8.5 uses top-level consumer declarations only for lineage and impact analysis. A consumer declaration says which governed semantic model, metrics, and dimensions a downstream report requires. It does **not** claim that a Power BI workspace, semantic model, report, refresh schedule, or deployment has already been provisioned. -Actual analytical delivery and Power BI service integration remain Phase 9 and Phase 10 responsibilities. +Phase 9 adds `products/` contracts that compose physical mart freshness, dbt build evidence, quality policy, semantic definitions, and declared consumers into an explicit publication-readiness gate. These contracts still do not represent Power BI deployment state; live Power BI service provisioning and refresh/deployment evidence remain Phase 10 responsibilities. diff --git a/contracts/analytics/products/README.md b/contracts/analytics/products/README.md new file mode 100644 index 0000000..86837d3 --- /dev/null +++ b/contracts/analytics/products/README.md @@ -0,0 +1,9 @@ +# Analytical product contracts + +Phase 9 source-controls the declaration of an analytical product separately from its runtime evidence. +A product contract names the curated source relation, dbt mart relation, semantic model, quality gate, +governed metrics, dimensions, and declared consumers required before Studio may call the product ready. + +These contracts do not represent Power BI deployment state. Phase 9 proves Studio analytical-product +readiness and publication boundaries; live Power BI workspace, report, refresh, and deployment evidence +remain Phase 10. diff --git a/contracts/analytics/products/fed_funds_rate_daily_product.v1.json b/contracts/analytics/products/fed_funds_rate_daily_product.v1.json new file mode 100644 index 0000000..7518f8f --- /dev/null +++ b/contracts/analytics/products/fed_funds_rate_daily_product.v1.json @@ -0,0 +1,30 @@ +{ + "code": "FED_FUNDS_RATE_DAILY_PRODUCT", + "version": "1.0.0", + "name": "Federal Funds Rate Daily Analytical Product", + "description": "Governed daily Federal Funds Rate analytical product published from Studio-curated data through dbt, semantic, quality, and consumer readiness gates.", + "owner": "SkyData Studio", + "domain": "Macroeconomics", + "source_relation": "mart.fed_funds_rate", + "mart_relation": "dbt_mart.fct_fed_funds_rate_daily", + "freshness_column": "observation_date", + "row_alignment": "EXACT", + "semantic_model": "fed_funds_rate_daily", + "quality_contract_code": "FED_FUNDS_RATE_DAILY_QUALITY", + "required_metrics": [ + "average_federal_funds_rate", + "federal_funds_observation_count", + "maximum_federal_funds_rate", + "minimum_federal_funds_rate" + ], + "required_dimensions": [ + "observation_date", + "observation_month", + "observation_year", + "rate_direction" + ], + "consumer_codes": [ + "FED_FUNDS_RATE_OVERVIEW" + ], + "publication_mode": "GATED" +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 9b98b5b..c42816c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -285,3 +285,22 @@ If Airflow is unavailable, the endpoint degrades to `PARTIAL` and preserves the Phase 8.5 closes the lineage chain at source-controlled analytics-consumer declarations. The declaration names the dbt semantic model, governed metrics, and dimensions that a report or application requires. Studio resolves those names against the generated dbt semantic artifacts at request time and emits metric β†’ consumer edges without creating another semantic or reporting metadata authority. The first proof declares a `Federal Funds Rate Overview` report that targets Power BI but has deployment status `DECLARED`. That status is intentional: Phase 8.5 proves downstream dependency and change-impact lineage only. It does not claim that a Power BI workspace, semantic model, report, refresh schedule, or deployment exists. Analytical product delivery remains Phase 9, while live Power BI service integration remains Phase 10. + +## Phase 9.1 analytical-product publication boundary + +Phase 9.1 turns the proven Federal Funds Rate mart into a governed analytical product without moving execution ownership into the Studio API. The analytical-product contract names its curated source relation, dbt mart relation, freshness column, quality contract, semantic model, governed metrics, and declared consumers. Studio composes those existing authorities into one publication-readiness read model. + +```text +mart.fed_funds_rate ── physical freshness ──┐ + β”‚ +dbt_mart.fct_fed_funds_rate_daily ─────────── + β”‚ +dbt build artifacts ───────────────────────── +quality contract ─────────────────────────────── publication gate +semantic model + metrics ──────────────────── +declared analytics consumers β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +Freshness is deliberately a blocking gate. A dbt model can remain structurally healthy while becoming stale after a newer Studio pipeline materialization advances the curated source. In that state the product reports `STALE`; Studio does not invoke dbt on the user's behalf. The operator runs the existing dbt build explicitly, after which source/mart row counts and latest freshness values can align and publication can become `READY`. + +No new database table is required for this proof. Product policy is source-controlled under `contracts/analytics/products/`; physical evidence is queried from Studio PostgreSQL; dbt artifacts, quality policy, semantic definitions, and consumer declarations remain owned by their existing layers. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index f813473..b93c485 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -703,7 +703,7 @@ Closure evidence: ## Phase 8.5 β€” Analytics Consumer Lineage and Impact Closure -**Status:** In progress. Consumer-contract resolution and metric-to-report impact are the active proof. +**Status:** Complete. Consumer-contract resolution and metric-to-report impact are proven. Scope: @@ -724,5 +724,40 @@ Acceptance proof: - selecting `average_federal_funds_rate` reports exactly one downstream consumer, `Federal Funds Rate Overview`; - the declaration remains `deployment_status=DECLARED`, proving lineage without fabricating Power BI deployment state; - the Lineage workbench renders the consumer dependency surface alongside the four completed Phase 8 evidence layers; -- local validation remains green with 92 pytest tests plus Ruff, mypy, npm audit, ESLint, and Vite build. +- local validation remains green with 92 pytest tests plus Ruff, mypy, ESLint, and Vite build. The same `npm ci` run surfaced one high-severity dependency advisory, which remains a separate dependency-security follow-up rather than being treated as clean audit evidence. + +Closure evidence: + +- `GET /api/v1/lineage/consumers/summary` returned `consumer_status=READY` and `semantic_artifact_status=READY`; +- one `FED_FUNDS_RATE_OVERVIEW` consumer resolved against all four declared governed metrics with zero unresolved metrics; +- the graph contained exactly 5 nodes and 4 `CONSUMED_BY` edges; +- metric impact for `average_federal_funds_rate` resolved exactly one downstream consumer, `Federal Funds Rate Overview`, with Power BI delivery status `DECLARED`; +- the Lineage workbench rendered consumers 1/1, metrics 4/4, semantic READY, and the declaration-only delivery boundary; +- final local validation passed Ruff, mypy across 60 source files, 92 pytest tests, ESLint, and Vite production build; `npm ci` surfaced one high-severity dependency advisory for explicit follow-up. + +## Phase 9.1 β€” Analytical Mart Publication Readiness and Freshness Gate + +**Status:** In progress. The first governed analytical product and publication-readiness gate are the active proof. + +Scope: + +- introduce a source-controlled analytical-product contract for the Federal Funds Rate daily product; +- preserve dbt as transformation authority, Phase 7 as quality-policy authority, and Phase 8.5 as consumer-dependency authority; +- read physical row-count and latest-date evidence from `mart.fed_funds_rate` and `dbt_mart.fct_fed_funds_rate_daily`; +- compose six publication gates: physical relations, dbt model build, freshness alignment, quality contract, semantic product, and declared consumers; +- expose `GET /api/v1/analytics/products/summary`; +- replace the Analytical Marts placeholder with a workbench that makes stale-vs-ready publication state explicit; +- never execute dbt silently from the API or UI; a stale mart instructs the operator to run the existing dbt build command explicitly; +- preserve Power BI as declaration-only downstream intent until live service provisioning is introduced later. + +Acceptance proof: + +- before a fresh dbt build, a curated source that has advanced beyond the analytical mart reports `product_status=STALE`, `freshness_status=STALE`, and a non-zero row-count/date delta; +- after `./scripts/dbt.ps1 build`, source and mart evidence align and `FRESHNESS_ALIGNMENT` passes; +- dbt mart build evidence is READY, the Federal Funds Rate quality contract is COMPLIANT, all four governed metrics resolve, and the declared report consumer resolves; +- all six publication gates pass and the analytical product reports `product_status=READY`; +- the Analytical Marts workbench renders source/mart evidence, row/date alignment, publication gates, and the explicit refresh boundary; +- local validation remains green with 96 pytest tests plus Ruff, mypy, ESLint, and Vite build; +- dependency-security follow-up remains explicit until the high-severity npm advisory surfaced during Phase 8.5 validation is identified and cleared. + diff --git a/docs/SkyData_RepoMap.md b/docs/SkyData_RepoMap.md index b5fa8ff..5ada0ce 100644 --- a/docs/SkyData_RepoMap.md +++ b/docs/SkyData_RepoMap.md @@ -18,6 +18,7 @@ SkyDataStudio/ β”‚ β”‚ β”‚ └── routes/ β”‚ β”‚ β”‚ β”œβ”€β”€ __init__.py β”‚ β”‚ β”‚ β”œβ”€β”€ airflow.py +β”‚ β”‚ β”‚ β”œβ”€β”€ analytics.py β”‚ β”‚ β”‚ β”œβ”€β”€ contracts.py β”‚ β”‚ β”‚ β”œβ”€β”€ dbt.py β”‚ β”‚ β”‚ β”œβ”€β”€ health.py @@ -53,6 +54,7 @@ SkyDataStudio/ β”‚ β”‚ β”œβ”€β”€ schemas/ β”‚ β”‚ β”‚ β”œβ”€β”€ __init__.py β”‚ β”‚ β”‚ β”œβ”€β”€ airflow.py +β”‚ β”‚ β”‚ β”œβ”€β”€ analytics.py β”‚ β”‚ β”‚ β”œβ”€β”€ assets.py β”‚ β”‚ β”‚ β”œβ”€β”€ dbt.py β”‚ β”‚ β”‚ β”œβ”€β”€ execution.py @@ -63,6 +65,7 @@ SkyDataStudio/ β”‚ β”‚ β”‚ └── quality.py β”‚ β”‚ └── services/ β”‚ β”‚ β”œβ”€β”€ __init__.py +β”‚ β”‚ β”œβ”€β”€ analytics_products.py β”‚ β”‚ β”œβ”€β”€ asset_detail.py β”‚ β”‚ β”œβ”€β”€ asset_workspace.py β”‚ β”‚ β”œβ”€β”€ contract_compatibility.py @@ -99,6 +102,7 @@ SkyDataStudio/ β”‚ β”‚ └── Topbar.jsx β”‚ β”œβ”€β”€ pages/ β”‚ β”‚ β”œβ”€β”€ Airflow.jsx +β”‚ β”‚ β”œβ”€β”€ AnalyticalMarts.jsx β”‚ β”‚ β”œβ”€β”€ DataAssets.jsx β”‚ β”‚ β”œβ”€β”€ DataContracts.jsx β”‚ β”‚ β”œβ”€β”€ DataModels.jsx @@ -120,7 +124,10 @@ SkyDataStudio/ β”œβ”€β”€ contracts/ β”‚ β”œβ”€β”€ analytics/ β”‚ β”‚ β”œβ”€β”€ fed_funds_rate_overview.v1.json -β”‚ β”‚ └── README.md +β”‚ β”‚ β”œβ”€β”€ README.md +β”‚ β”‚ └── products/ +β”‚ β”‚ β”œβ”€β”€ fed_funds_rate_daily_product.v1.json +β”‚ β”‚ └── README.md β”‚ β”œβ”€β”€ quality/ β”‚ β”‚ β”œβ”€β”€ fed_funds_rate_daily.v1.json β”‚ β”‚ └── README.md @@ -167,6 +174,7 @@ SkyDataStudio/ β”‚ β”œβ”€β”€ conftest.py β”‚ β”œβ”€β”€ test_airflow_api.py β”‚ β”œβ”€β”€ test_airflow_client.py +β”‚ β”œβ”€β”€ test_analytics_products.py β”‚ β”œβ”€β”€ test_asset_detail.py β”‚ β”œβ”€β”€ test_asset_workspace_api.py β”‚ β”œβ”€β”€ test_asset_workspace.py diff --git a/docs/repository-map.txt b/docs/repository-map.txt index 40be159..4109db0 100644 --- a/docs/repository-map.txt +++ b/docs/repository-map.txt @@ -9,6 +9,7 @@ SkyDataStudio β”‚ β”‚ β”‚ β”œβ”€β”€ routes β”‚ β”‚ β”‚ β”‚ β”œβ”€β”€ __init__.py β”‚ β”‚ β”‚ β”‚ β”œβ”€β”€ airflow.py +β”‚ β”‚ β”‚ β”‚ β”œβ”€β”€ analytics.py β”‚ β”‚ β”‚ β”‚ β”œβ”€β”€ contracts.py β”‚ β”‚ β”‚ β”‚ β”œβ”€β”€ dbt.py β”‚ β”‚ β”‚ β”‚ β”œβ”€β”€ health.py @@ -46,6 +47,7 @@ SkyDataStudio β”‚ β”‚ β”œβ”€β”€ schemas β”‚ β”‚ β”‚ β”œβ”€β”€ __init__.py β”‚ β”‚ β”‚ β”œβ”€β”€ airflow.py +β”‚ β”‚ β”‚ β”œβ”€β”€ analytics.py β”‚ β”‚ β”‚ β”œβ”€β”€ assets.py β”‚ β”‚ β”‚ β”œβ”€β”€ dbt.py β”‚ β”‚ β”‚ β”œβ”€β”€ execution.py @@ -56,6 +58,7 @@ SkyDataStudio β”‚ β”‚ β”‚ └── quality.py β”‚ β”‚ β”œβ”€β”€ services β”‚ β”‚ β”‚ β”œβ”€β”€ __init__.py +β”‚ β”‚ β”‚ β”œβ”€β”€ analytics_products.py β”‚ β”‚ β”‚ β”œβ”€β”€ asset_detail.py β”‚ β”‚ β”‚ β”œβ”€β”€ asset_workspace.py β”‚ β”‚ β”‚ β”œβ”€β”€ contract_compatibility.py @@ -86,6 +89,7 @@ SkyDataStudio β”‚ β”‚ β”‚ └── Topbar.jsx β”‚ β”‚ β”œβ”€β”€ pages β”‚ β”‚ β”‚ β”œβ”€β”€ Airflow.jsx +β”‚ β”‚ β”‚ β”œβ”€β”€ AnalyticalMarts.jsx β”‚ β”‚ β”‚ β”œβ”€β”€ DataAssets.jsx β”‚ β”‚ β”‚ β”œβ”€β”€ DataContracts.jsx β”‚ β”‚ β”‚ β”œβ”€β”€ DataModels.jsx @@ -114,6 +118,9 @@ SkyDataStudio β”‚ └── vite.config.js β”œβ”€β”€ contracts β”‚ β”œβ”€β”€ analytics +β”‚ β”‚ β”œβ”€β”€ products +β”‚ β”‚ β”‚ β”œβ”€β”€ fed_funds_rate_daily_product.v1.json +β”‚ β”‚ β”‚ └── README.md β”‚ β”‚ β”œβ”€β”€ fed_funds_rate_overview.v1.json β”‚ β”‚ └── README.md β”‚ β”œβ”€β”€ quality @@ -162,6 +169,7 @@ SkyDataStudio β”‚ β”œβ”€β”€ conftest.py β”‚ β”œβ”€β”€ test_airflow_api.py β”‚ β”œβ”€β”€ test_airflow_client.py +β”‚ β”œβ”€β”€ test_analytics_products.py β”‚ β”œβ”€β”€ test_asset_detail.py β”‚ β”œβ”€β”€ test_asset_workspace.py β”‚ β”œβ”€β”€ test_asset_workspace_api.py diff --git a/tests/test_analytics_products.py b/tests/test_analytics_products.py new file mode 100644 index 0000000..073fe4d --- /dev/null +++ b/tests/test_analytics_products.py @@ -0,0 +1,297 @@ +from collections.abc import Generator +from typing import Literal + +import pytest +from fastapi.testclient import TestClient +from skydata_studio.api.routes import analytics as analytics_route +from skydata_studio.db.session import get_session +from skydata_studio.main import app +from skydata_studio.schemas.analytics import ( + AnalyticsProductDefinition, + AnalyticsProductSummary, + AnalyticsRelationEvidence, +) +from skydata_studio.schemas.dbt import ( + DbtModelCatalogueItem, + DbtModelCatalogueSummary, + DbtSemanticDimensionSummary, + DbtSemanticLayerSummary, + DbtSemanticMetricSummary, + DbtSemanticModelSummary, +) +from skydata_studio.schemas.lineage import ( + AnalyticsConsumerDefinition, + AnalyticsConsumerImpactSummary, + AnalyticsConsumerLineageSummary, + AnalyticsConsumerNode, +) +from skydata_studio.schemas.quality import QualityContractSummary +from skydata_studio.services.analytics_products import compose_analytics_product_summary +from sqlalchemy.orm import Session + +client = TestClient(app) + + +def _product() -> AnalyticsProductDefinition: + return AnalyticsProductDefinition( + code="FED_FUNDS_RATE_DAILY_PRODUCT", + version="1.0.0", + name="Federal Funds Rate Daily Analytical Product", + description="Proof product.", + owner="SkyData Studio", + domain="Macroeconomics", + source_relation="mart.fed_funds_rate", + mart_relation="dbt_mart.fct_fed_funds_rate_daily", + freshness_column="observation_date", + semantic_model="fed_funds_rate_daily", + quality_contract_code="FED_FUNDS_RATE_DAILY_QUALITY", + required_metrics=[ + "average_federal_funds_rate", + "federal_funds_observation_count", + "maximum_federal_funds_rate", + "minimum_federal_funds_rate", + ], + required_dimensions=[ + "observation_date", + "observation_month", + "observation_year", + "rate_direction", + ], + consumer_codes=["FED_FUNDS_RATE_OVERVIEW"], + ) + + +def _relation(relation: str, rows: int, latest: str) -> AnalyticsRelationEvidence: + return AnalyticsRelationEvidence( + relation=relation, + status="READY", + row_count=rows, + max_freshness_value=latest, + ) + + +def _catalogue( + status: Literal["READY", "ERROR", "UNKNOWN"] = "READY", +) -> DbtModelCatalogueSummary: + model = DbtModelCatalogueItem( + unique_id="model.skydata_studio.fct_fed_funds_rate_daily", + name="fct_fed_funds_rate_daily", + layer="MART", + relation="dbt_mart.fct_fed_funds_rate_daily", + materialization="TABLE", + build_status=status, + path="models/marts/fct_fed_funds_rate_daily.sql", + tags=[], + columns=[], + upstream=[], + downstream=[], + test_count=5, + ) + return DbtModelCatalogueSummary( + artifact_status="READY", + model_count=1, + ready_model_count=1 if status == "READY" else 0, + source_count=1, + test_count=5, + models=[model], + ) + + +def _semantic() -> DbtSemanticLayerSummary: + dimensions = [ + DbtSemanticDimensionSummary( + name="observation_date", dimension_type="TIME", granularity="DAY" + ), + DbtSemanticDimensionSummary( + name="observation_month", dimension_type="TIME", granularity="MONTH" + ), + DbtSemanticDimensionSummary( + name="observation_year", dimension_type="CATEGORICAL" + ), + DbtSemanticDimensionSummary(name="rate_direction", dimension_type="CATEGORICAL"), + ] + metric_names = [ + "average_federal_funds_rate", + "federal_funds_observation_count", + "maximum_federal_funds_rate", + "minimum_federal_funds_rate", + ] + metrics = [ + DbtSemanticMetricSummary( + unique_id=f"metric.skydata_studio.{name}", + name=name, + label=name.replace("_", " ").title(), + metric_type="SIMPLE", + semantic_model="fed_funds_rate_daily", + ) + for name in metric_names + ] + model = DbtSemanticModelSummary( + unique_id="semantic_model.skydata_studio.fed_funds_rate_daily", + name="fed_funds_rate_daily", + relation="dbt_mart.fct_fed_funds_rate_daily", + entities=[], + dimensions=dimensions, + metric_names=metric_names, + ) + return DbtSemanticLayerSummary( + artifact_status="READY", + semantic_model_count=1, + metric_count=4, + entity_count=0, + dimension_count=4, + semantic_models=[model], + metrics=metrics, + ) + + +def _quality( + status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"] = "COMPLIANT", +) -> QualityContractSummary: + return QualityContractSummary( + contract_code="FED_FUNDS_RATE_DAILY_QUALITY", + contract_version="1.1.0", + contract_name="Federal Funds Rate Daily Quality Contract", + description="Proof quality contract.", + target_name="fct_fed_funds_rate_daily", + layer="MART", + enforcement_mode="BLOCK", + artifact_status="READY", + evidence_trust_posture="TRUSTED", + contract_status=status, + minimum_pass_rate=1.0, + pass_rate=1.0 if status == "COMPLIANT" else 0.8, + required_rule_count=5, + satisfied_rule_count=5 if status == "COMPLIANT" else 4, + warning_rule_count=0, + blocking_rule_count=0 if status == "COMPLIANT" else 1, + missing_rule_count=0, + source_path="contracts/quality/fed_funds_rate_daily.v1.json", + rules=[], + ) + + +def _consumers() -> AnalyticsConsumerLineageSummary: + definition = AnalyticsConsumerDefinition( + code="FED_FUNDS_RATE_OVERVIEW", + version="1.0.0", + name="Federal Funds Rate Overview", + description="Proof consumer.", + consumer_type="REPORT", + delivery_system="POWER_BI", + deployment_status="DECLARED", + semantic_model="fed_funds_rate_daily", + required_metrics=[ + "average_federal_funds_rate", + "federal_funds_observation_count", + "maximum_federal_funds_rate", + "minimum_federal_funds_rate", + ], + required_dimensions=[ + "observation_date", + "observation_month", + "observation_year", + "rate_direction", + ], + owner="SkyData Studio", + ) + node = AnalyticsConsumerNode( + id="consumer:fed_funds_rate_overview", + label="Federal Funds Rate Overview", + node_type="ANALYTICS_CONSUMER", + system="POWER_BI", + status="READY", + metadata={"consumer_code": "FED_FUNDS_RATE_OVERVIEW"}, + ) + return AnalyticsConsumerLineageSummary( + consumer_status="READY", + semantic_artifact_status="READY", + consumer_contract_count=1, + resolved_consumer_count=1, + declared_metric_count=4, + resolved_metric_count=4, + unresolved_metric_count=0, + node_count=1, + edge_count=0, + source_paths=[], + consumers=[definition], + metric_bindings=[], + nodes=[node], + edges=[], + default_impact=AnalyticsConsumerImpactSummary(), + ) + + +def _summary( + *, + source_rows: int = 26340, + mart_rows: int = 26340, + quality_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"] = "COMPLIANT", +) -> AnalyticsProductSummary: + return compose_analytics_product_summary( + product=_product(), + source_path="contracts/analytics/products/fed_funds_rate_daily_product.v1.json", + source=_relation("mart.fed_funds_rate", source_rows, "2026-08-13"), + mart=_relation("dbt_mart.fct_fed_funds_rate_daily", mart_rows, "2026-08-13"), + catalogue=_catalogue(), + semantic=_semantic(), + quality=_quality(quality_status), + consumers=_consumers(), + ) + + +def test_analytical_product_is_ready_when_all_publication_gates_pass() -> None: + summary = _summary() + + assert summary.product_status == "READY" + assert summary.freshness_status == "ALIGNED" + assert summary.row_count_delta == 0 + assert summary.resolved_metric_count == 4 + assert summary.resolved_consumer_count == 1 + assert all(gate.status == "PASS" for gate in summary.gates) + + +def test_analytical_product_is_stale_when_curated_source_is_ahead() -> None: + summary = _summary(source_rows=26340, mart_rows=26335) + + assert summary.product_status == "STALE" + assert summary.freshness_status == "STALE" + assert summary.refresh_required is True + assert summary.row_count_delta == 5 + freshness_gate = next(gate for gate in summary.gates if gate.code == "FRESHNESS_ALIGNMENT") + assert freshness_gate.status == "BLOCK" + + +def test_analytical_product_is_blocked_by_noncompliant_quality_contract() -> None: + summary = _summary(quality_status="BLOCKED") + + assert summary.product_status == "BLOCKED" + quality_gate = next(gate for gate in summary.gates if gate.code == "QUALITY_CONTRACT") + assert quality_gate.status == "BLOCK" + + +def test_analytics_product_endpoint_projects_service_summary( + monkeypatch: pytest.MonkeyPatch, +) -> None: + expected = _summary() + + def fake_summary(session: Session) -> AnalyticsProductSummary: + del session + return expected + + def override_session() -> Generator[Session, None, None]: + with Session() as session: + yield session + + monkeypatch.setattr(analytics_route, "analytics_product_summary", fake_summary) + app.dependency_overrides[get_session] = override_session + try: + response = client.get("/api/v1/analytics/products/summary") + finally: + app.dependency_overrides.pop(get_session, None) + + assert response.status_code == 200 + payload = response.json() + assert payload["product_status"] == "READY" + assert payload["freshness_status"] == "ALIGNED" + assert payload["resolved_metric_count"] == 4 diff --git a/tests/test_platform.py b/tests/test_platform.py index ac0ad60..421e069 100644 --- a/tests/test_platform.py +++ b/tests/test_platform.py @@ -13,7 +13,7 @@ def test_platform_summary_exposes_product_boundary() -> None: assert payload["theme"] == "Aurora Foundry" assert "SkyCommand" in payload["boundary"] assert payload["current_phase"] == ( - "Phase 8.5 β€” Analytics Consumer Lineage and Impact Closure" + "Phase 9.1 β€” Analytical Mart Publication Readiness and Freshness Gate" ) assert {item["code"] for item in payload["capabilities"]} >= { "AIRFLOW",