Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -519,21 +519,21 @@ The repository currently includes:
- a completed Phase 8.1 asset/model/semantic/metric lineage graph with transitive impact analysis;
- 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 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 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;
- backend/frontend validation, repository map, and compact handoff tooling.

The active implementation target is **Phase 8.4 — Pipeline and Airflow Execution Lineage Foundation**. Phases 0 through 7 are complete; Phase 8.1 established asset-level impact, Phase 8.2 added field-level impact, and Phase 8.3 projected trust evidence onto those nodes. Phase 8.4 now joins the latest Airflow DAG/task proof to the replay-safe Studio pipeline run and its structured materialization steps without reading Airflow metadata tables directly.
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.


## Phase 8.4 quick start
## Phase 8.5 quick start

Phase 8.4 federates runtime evidence into lineage without moving execution ownership. Airflow remains authoritative for DAG-run and task-instance state; Studio remains authoritative for pipeline-run and step evidence. The runtime graph links the existing `DFF` and `FED_FUNDS_RATE_MART` structural nodes to the latest `AIRFLOW:` Studio run.
Phase 8.5 resolves declared analytics dependencies against dbt-owned semantic artifacts. The first proof declares a Federal Funds Rate Overview report that requires all four governed Federal Funds Rate metrics. The report targets Power BI but remains `DECLARED`; no Power BI deployment is claimed.

```powershell
# Ensure the Airflow proof runtime is available, then start the Studio API/web as usual.
Invoke-RestMethod `
"http://localhost:8100/api/v1/lineage/runtime/summary" |
"http://localhost:8100/api/v1/lineage/consumers/summary" |
ConvertTo-Json -Depth 12
```

A fully connected proof reports `runtime_status=READY`, four successful Airflow tasks, four successful Studio steps, and replay-safe materialization evidence for `mart.fed_funds_rate`.
A complete proof reports one resolved consumer, four resolved metrics, zero unresolved dependencies, 5 nodes, and 4 `CONSUMED_BY` edges.
26 changes: 26 additions & 0 deletions apps/api/skydata_studio/api/routes/lineage.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
from skydata_studio.core.config import Settings, get_settings
from skydata_studio.db.session import SessionDependency
from skydata_studio.schemas.lineage import (
AnalyticsConsumerImpactSummary,
AnalyticsConsumerLineageSummary,
FieldLineageImpactSummary,
FieldLineageSummary,
LineageImpactSummary,
Expand All @@ -19,6 +21,10 @@
lineage_impact,
lineage_summary,
)
from skydata_studio.services.lineage_consumers import (
analytics_consumer_impact,
analytics_consumer_lineage_summary,
)
from skydata_studio.services.lineage_runtime import runtime_lineage_summary
from skydata_studio.services.lineage_trust import lineage_trust_summary

Expand Down Expand Up @@ -96,3 +102,23 @@ def lineage_runtime_execution(
return runtime_lineage_summary(session, settings)
except (SQLAlchemyError, OSError, ValueError, TypeError) as error:
raise _unavailable(error) from error


@router.get("/consumers/summary", response_model=AnalyticsConsumerLineageSummary)
def lineage_analytics_consumers(
focus_metric_name: Annotated[str | None, Query(alias="focusMetricName")] = None,
) -> AnalyticsConsumerLineageSummary:
try:
return analytics_consumer_lineage_summary(focus_metric_name=focus_metric_name)
except (OSError, ValueError, TypeError) as error:
raise _unavailable(error) from error


@router.get("/consumers/impact", response_model=AnalyticsConsumerImpactSummary)
def lineage_analytics_consumer_impact(
metric_name: Annotated[str, Query(alias="metricName", min_length=1)],
) -> AnalyticsConsumerImpactSummary:
try:
return analytics_consumer_impact(metric_name)
except (OSError, ValueError, TypeError) as error:
raise _unavailable(error) from error
2 changes: 1 addition & 1 deletion apps/api/skydata_studio/api/routes/platform.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ def platform_summary() -> PlatformSummary:
product="SkyData Studio",
subtitle="Data Engineering Workbench",
theme="Aurora Foundry",
current_phase="Phase 8.4 — Pipeline and Airflow Execution Lineage Foundation",
current_phase="Phase 8.5 — Analytics Consumer Lineage and Impact Closure",
boundary=(
"SkyData Studio starts after SkyCommand ingestion and publishes governed "
"analytical products for SkyWeb Analytics, Power BI, and future consumers."
Expand Down
71 changes: 71 additions & 0 deletions apps/api/skydata_studio/schemas/lineage.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,3 +241,74 @@ class RuntimeLineageSummary(BaseModel):
nodes: list[RuntimeLineageNode] = Field(default_factory=list)
edges: list[RuntimeLineageEdge] = Field(default_factory=list)


AnalyticsConsumerType = Literal["REPORT", "APPLICATION"]
AnalyticsConsumerDeploymentStatus = Literal["DECLARED", "DEPLOYED", "RETIRED"]
AnalyticsConsumerResolutionStatus = Literal["READY", "PARTIAL", "MISSING"]


class AnalyticsConsumerDefinition(BaseModel):
code: str
version: str
name: str
description: str
consumer_type: AnalyticsConsumerType
delivery_system: str
deployment_status: AnalyticsConsumerDeploymentStatus
semantic_model: str
required_metrics: list[str]
required_dimensions: list[str] = Field(default_factory=list)
owner: str


class AnalyticsConsumerMetricBinding(BaseModel):
consumer_code: str
metric_name: str
metric_label: str | None = None
metric_unique_id: str | None = None
resolved: bool
message: str


class AnalyticsConsumerNode(BaseModel):
id: str
label: str
node_type: Literal["METRIC", "ANALYTICS_CONSUMER"]
system: str
status: str
metadata: dict[str, object] = Field(default_factory=dict)


class AnalyticsConsumerEdge(BaseModel):
id: str
upstream_id: str
downstream_id: str
edge_type: Literal["CONSUMED_BY"]
label: str


class AnalyticsConsumerImpactSummary(BaseModel):
selected_metric_name: str | None = None
selected_metric_label: str | None = None
downstream_consumer_count: int = Field(default=0, ge=0)
consumers: list[AnalyticsConsumerNode] = Field(default_factory=list)


class AnalyticsConsumerLineageSummary(BaseModel):
project_name: str = "skydata_studio"
phase: str = "Phase 8.5 — Analytics Consumer Lineage and Impact Closure"
consumer_status: AnalyticsConsumerResolutionStatus
semantic_artifact_status: Literal["READY", "PENDING", "MISSING"]
consumer_contract_count: int = Field(default=0, ge=0)
resolved_consumer_count: int = Field(default=0, ge=0)
declared_metric_count: int = Field(default=0, ge=0)
resolved_metric_count: int = Field(default=0, ge=0)
unresolved_metric_count: int = Field(default=0, ge=0)
node_count: int = Field(default=0, ge=0)
edge_count: int = Field(default=0, ge=0)
source_paths: list[str] = Field(default_factory=list)
consumers: list[AnalyticsConsumerDefinition] = Field(default_factory=list)
metric_bindings: list[AnalyticsConsumerMetricBinding] = Field(default_factory=list)
nodes: list[AnalyticsConsumerNode] = Field(default_factory=list)
edges: list[AnalyticsConsumerEdge] = Field(default_factory=list)
default_impact: AnalyticsConsumerImpactSummary
Loading