981 lines
32 KiB
Python
981 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections.abc import Mapping
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.audit.logging import audit_event
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
|
|
from govoplan_core.core.datasources import (
|
|
DatasourceAccessError,
|
|
DatasourceDescriptor,
|
|
DatasourceError,
|
|
DatasourceGovernance,
|
|
DatasourceMaterialization,
|
|
DatasourceNotFoundError,
|
|
DatasourceOrigin,
|
|
DatasourceReadRequest,
|
|
DatasourceStage,
|
|
DatasourceStageInput,
|
|
DatasourceUnavailableError,
|
|
datasource_origins,
|
|
)
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_datasources.backend.runtime import get_registry
|
|
from govoplan_datasources.backend.schemas import (
|
|
DatasourceFieldResponse,
|
|
DatasourceFreezeRequest,
|
|
DatasourceGovernancePayload,
|
|
DatasourceGovernanceUpdateRequest,
|
|
DatasourceLifecycleEvidenceListResponse,
|
|
DatasourceLifecycleEvidenceResponse,
|
|
DatasourceListResponse,
|
|
DatasourceMaterializationListResponse,
|
|
DatasourceMaterializationResponse,
|
|
DatasourceOriginHealthResponse,
|
|
DatasourceOriginListResponse,
|
|
DatasourceOriginPushdownResponse,
|
|
DatasourceOriginRegisterRequest,
|
|
DatasourceOriginResponse,
|
|
DatasourcePreviewDiagnosticResponse,
|
|
DatasourcePreviewResponse,
|
|
DatasourceResponse,
|
|
DatasourceRetentionApplyRequest,
|
|
DatasourceRetentionApplyResponse,
|
|
DatasourceRetentionCandidateResponse,
|
|
DatasourceRetentionPlanResponse,
|
|
DatasourceRetireResponse,
|
|
DatasourceStageCreateRequest,
|
|
DatasourceStageDecisionRequest,
|
|
DatasourceStageListResponse,
|
|
DatasourceStagePromoteRequest,
|
|
DatasourceStagePromoteResponse,
|
|
DatasourceStageResponse,
|
|
)
|
|
from govoplan_datasources.backend.service import (
|
|
ADMIN_SCOPE,
|
|
CATALOGUE_READ_SCOPE,
|
|
SOURCE_WRITE_SCOPE,
|
|
STAGE_APPROVE_SCOPE,
|
|
STAGE_WRITE_SCOPE,
|
|
SqlDatasourceProvider,
|
|
)
|
|
from govoplan_datasources.backend.tabular import parse_csv_rows
|
|
|
|
|
|
router = APIRouter(prefix="/datasources", tags=["datasources"])
|
|
|
|
|
|
def _provider() -> SqlDatasourceProvider:
|
|
return SqlDatasourceProvider(registry=get_registry())
|
|
|
|
|
|
def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None:
|
|
if any(has_scope(principal, scope) for scope in scopes):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Missing one of the required scopes: {', '.join(scopes)}",
|
|
)
|
|
|
|
|
|
def _http_error(exc: DatasourceError) -> HTTPException:
|
|
if isinstance(exc, DatasourceNotFoundError):
|
|
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
|
|
if isinstance(exc, DatasourceAccessError):
|
|
return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
|
|
if isinstance(exc, DatasourceUnavailableError):
|
|
return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
|
|
return HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail=str(exc),
|
|
)
|
|
|
|
|
|
@router.get("/origins", response_model=DatasourceOriginListResponse)
|
|
def api_list_origins(
|
|
query: str = Query(default="", max_length=200),
|
|
limit: int = Query(default=100, ge=1, le=100),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceOriginListResponse:
|
|
_require_any_scope(
|
|
principal,
|
|
CATALOGUE_READ_SCOPE,
|
|
SOURCE_WRITE_SCOPE,
|
|
ADMIN_SCOPE,
|
|
)
|
|
provider = datasource_origins(get_registry())
|
|
if provider is None:
|
|
return DatasourceOriginListResponse(available=False, origins=[])
|
|
try:
|
|
origins = provider.list_origins(
|
|
session,
|
|
principal,
|
|
query=query,
|
|
limit=limit,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourceOriginListResponse(
|
|
available=True,
|
|
origins=[_origin_response(origin) for origin in origins],
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/origins/register",
|
|
response_model=DatasourceResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def api_register_origin(
|
|
payload: DatasourceOriginRegisterRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceResponse:
|
|
_require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
datasource = _provider().register_origin(
|
|
session,
|
|
principal,
|
|
origin_ref=payload.origin_ref,
|
|
name=payload.name,
|
|
source_name=payload.source_name,
|
|
mode=payload.mode,
|
|
description=payload.description,
|
|
governance=_governance(payload.governance),
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.origin.registered",
|
|
object_type="datasource",
|
|
object_id=datasource.ref,
|
|
details={
|
|
"origin_ref": payload.origin_ref,
|
|
"mode": datasource.mode,
|
|
"source_name": datasource.source_name,
|
|
},
|
|
)
|
|
session.commit()
|
|
return _datasource_response(datasource)
|
|
|
|
|
|
@router.get("/stages", response_model=DatasourceStageListResponse)
|
|
def api_list_stages(
|
|
limit: int = Query(default=100, ge=1, le=100),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceStageListResponse:
|
|
_require_any_scope(
|
|
principal,
|
|
CATALOGUE_READ_SCOPE,
|
|
STAGE_WRITE_SCOPE,
|
|
ADMIN_SCOPE,
|
|
)
|
|
try:
|
|
stages = _provider().list_stages(session, principal, limit=limit)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourceStageListResponse(
|
|
stages=[_stage_response(stage) for stage in stages]
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/stages",
|
|
response_model=DatasourceStageResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def api_create_stage(
|
|
payload: DatasourceStageCreateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceStageResponse:
|
|
_require_any_scope(principal, STAGE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
rows = (
|
|
tuple(payload.rows or ())
|
|
if payload.format == "json"
|
|
else parse_csv_rows(payload.csv_text or "", delimiter=payload.delimiter)
|
|
)
|
|
stage = _provider().create_stage(
|
|
session,
|
|
principal,
|
|
stage=DatasourceStageInput(
|
|
name=payload.name,
|
|
source_name=payload.source_name,
|
|
description=payload.description,
|
|
kind=payload.kind,
|
|
mode=payload.mode,
|
|
shape=payload.shape,
|
|
rows=rows,
|
|
target_datasource_ref=payload.target_datasource_ref,
|
|
provider="datasources.upload",
|
|
provenance={
|
|
**payload.provenance,
|
|
"created_via": "datasources",
|
|
"source_format": payload.format,
|
|
},
|
|
metadata=payload.metadata,
|
|
governance=_governance(payload.governance),
|
|
),
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.stage.created",
|
|
object_type="datasource_stage",
|
|
object_id=stage.ref,
|
|
details={
|
|
"source_name": stage.source_name,
|
|
"mode": stage.mode,
|
|
"row_count": stage.row_count,
|
|
"fingerprint": stage.fingerprint,
|
|
"validation_valid": stage.validation.get("valid"),
|
|
"quality_policy_hash": stage.validation.get("policy_hash"),
|
|
"schema_classification": _safe_mapping(
|
|
stage.validation.get("schema_change")
|
|
).get("classification"),
|
|
"approval_state": stage.approval.get("state"),
|
|
"approval_policy_hash": stage.approval.get("policy_hash"),
|
|
},
|
|
)
|
|
session.commit()
|
|
return _stage_response(stage)
|
|
|
|
|
|
@router.post(
|
|
"/stages/{stage_id}/decision",
|
|
response_model=DatasourceStageResponse,
|
|
)
|
|
def api_decide_stage(
|
|
stage_id: str,
|
|
payload: DatasourceStageDecisionRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceStageResponse:
|
|
_require_any_scope(principal, STAGE_APPROVE_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
stage, evidence_hash, replayed = _provider().decide_stage(
|
|
session,
|
|
principal,
|
|
stage_ref=f"stage:{stage_id}",
|
|
decision=payload.decision,
|
|
reason=payload.reason,
|
|
expected_policy_hash=payload.expected_policy_hash,
|
|
expected_subject_digest=payload.expected_subject_digest,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
action = (
|
|
"datasources.stage.approved"
|
|
if payload.decision == "approve"
|
|
else "datasources.stage.rejected"
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action=action,
|
|
object_type="datasource_stage",
|
|
object_id=stage.ref,
|
|
details={
|
|
"decision": payload.decision,
|
|
"resulting_state": stage.state,
|
|
"approval_count": stage.approval.get("approval_count"),
|
|
"required_approvals": stage.approval.get("required_approvals"),
|
|
"policy_hash": stage.approval.get("policy_hash"),
|
|
"evidence_hash": evidence_hash,
|
|
"replayed": replayed,
|
|
},
|
|
)
|
|
session.commit()
|
|
return _stage_response(stage)
|
|
|
|
|
|
@router.post(
|
|
"/stages/{stage_id}/promote",
|
|
response_model=DatasourceStagePromoteResponse,
|
|
)
|
|
def api_promote_stage(
|
|
stage_id: str,
|
|
payload: DatasourceStagePromoteRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceStagePromoteResponse:
|
|
_require_any_scope(principal, STAGE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
datasource, materialization = _provider().promote_stage(
|
|
session,
|
|
principal,
|
|
stage_ref=f"stage:{stage_id}",
|
|
freeze=payload.freeze,
|
|
frozen_label=payload.frozen_label,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.stage.promoted",
|
|
object_type="datasource",
|
|
object_id=datasource.ref,
|
|
details={
|
|
"stage_ref": f"stage:{stage_id}",
|
|
"materialization_ref": materialization.ref,
|
|
"revision": materialization.revision,
|
|
"frozen": materialization.frozen_at is not None,
|
|
"quality_policy_hash": _safe_mapping(
|
|
materialization.provenance.get("stage_validation")
|
|
).get("policy_hash"),
|
|
"approval_policy_hash": _safe_mapping(
|
|
materialization.provenance.get("stage_approval")
|
|
).get("policy_hash"),
|
|
"promotion_evidence_hash": materialization.provenance.get(
|
|
"promotion_evidence_hash"
|
|
),
|
|
},
|
|
)
|
|
session.commit()
|
|
return DatasourceStagePromoteResponse(
|
|
datasource=_datasource_response(datasource),
|
|
materialization=_materialization_response(materialization),
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/lifecycle-evidence",
|
|
response_model=DatasourceLifecycleEvidenceListResponse,
|
|
)
|
|
def api_list_lifecycle_evidence(
|
|
subject_ref: str | None = Query(default=None, max_length=160),
|
|
limit: int = Query(default=200, ge=1, le=500),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceLifecycleEvidenceListResponse:
|
|
_require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
rows = _provider().list_lifecycle_evidence(
|
|
session,
|
|
principal,
|
|
subject_ref=subject_ref,
|
|
limit=limit,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourceLifecycleEvidenceListResponse(
|
|
evidence=[_evidence_response(item) for item in rows]
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/retention/plan",
|
|
response_model=DatasourceRetentionPlanResponse,
|
|
)
|
|
def api_preview_retention(
|
|
as_of: datetime | None = Query(default=None),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceRetentionPlanResponse:
|
|
_require_any_scope(principal, ADMIN_SCOPE)
|
|
effective_at = as_of or datetime.now(UTC)
|
|
try:
|
|
plan = _provider().preview_retention(
|
|
session,
|
|
principal,
|
|
as_of=effective_at,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourceRetentionPlanResponse(
|
|
as_of=plan.as_of.isoformat(),
|
|
plan_hash=plan.plan_hash,
|
|
candidates=[
|
|
DatasourceRetentionCandidateResponse(**item.to_dict())
|
|
for item in plan.candidates
|
|
],
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/retention/apply",
|
|
response_model=DatasourceRetentionApplyResponse,
|
|
)
|
|
def api_apply_retention(
|
|
payload: DatasourceRetentionApplyRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceRetentionApplyResponse:
|
|
_require_any_scope(principal, ADMIN_SCOPE)
|
|
try:
|
|
disposed, evidence_hashes = _provider().apply_retention(
|
|
session,
|
|
principal,
|
|
as_of=payload.as_of,
|
|
plan_hash=payload.plan_hash,
|
|
target_refs=payload.target_refs,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.retention.applied",
|
|
object_type="datasource_retention_plan",
|
|
object_id=payload.plan_hash,
|
|
details={
|
|
"as_of": payload.as_of.isoformat(),
|
|
"disposed_refs": list(disposed),
|
|
"evidence_hashes": list(evidence_hashes),
|
|
},
|
|
)
|
|
session.commit()
|
|
return DatasourceRetentionApplyResponse(
|
|
plan_hash=payload.plan_hash,
|
|
disposed_refs=list(disposed),
|
|
evidence_hashes=list(evidence_hashes),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=DatasourceListResponse)
|
|
def api_list_datasources(
|
|
query: str = Query(default="", max_length=200),
|
|
limit: int = Query(default=100, ge=1, le=100),
|
|
authority_mode: str | None = Query(default=None, max_length=40),
|
|
classification: str | None = Query(default=None, max_length=80),
|
|
publication_state: str | None = Query(default=None, max_length=50),
|
|
owner_ref: str | None = Query(default=None, max_length=500),
|
|
responsible_organization_ref: str | None = Query(default=None, max_length=500),
|
|
affected_ref: str | None = Query(default=None, max_length=1000),
|
|
dependency_ref: str | None = Query(default=None, max_length=1000),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceListResponse:
|
|
_require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
items = _provider().list_datasources(
|
|
session,
|
|
principal,
|
|
query=query,
|
|
limit=limit,
|
|
authority_mode=authority_mode,
|
|
classification=classification,
|
|
publication_state=publication_state,
|
|
owner_ref=owner_ref,
|
|
responsible_organization_ref=responsible_organization_ref,
|
|
affected_ref=affected_ref,
|
|
dependency_ref=dependency_ref,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourceListResponse(
|
|
datasources=[_datasource_response(item) for item in items]
|
|
)
|
|
|
|
|
|
@router.get("/{datasource_id}", response_model=DatasourceResponse)
|
|
def api_get_datasource(
|
|
datasource_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceResponse:
|
|
_require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
item = _provider().get_datasource(
|
|
session,
|
|
principal,
|
|
datasource_ref=f"datasource:{datasource_id}",
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
if item is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Datasource not found.",
|
|
)
|
|
return _datasource_response(item)
|
|
|
|
|
|
@router.patch(
|
|
"/{datasource_id}/governance",
|
|
response_model=DatasourceResponse,
|
|
)
|
|
def api_update_datasource_governance(
|
|
datasource_id: str,
|
|
payload: DatasourceGovernanceUpdateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceResponse:
|
|
_require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
item = _provider().update_datasource_governance(
|
|
session,
|
|
principal,
|
|
datasource_ref=f"datasource:{datasource_id}",
|
|
governance=_governance(payload.governance)
|
|
or DatasourceGovernance(),
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.governance.updated",
|
|
object_type="datasource",
|
|
object_id=item.ref,
|
|
details={
|
|
"classification": item.governance.classification,
|
|
"publication_state": item.governance.publication_state,
|
|
"access_policy_ref": item.governance.access_policy_ref,
|
|
"visibility_policy_configured": bool(
|
|
item.governance.visibility_policy
|
|
),
|
|
"visibility_policy_hash": _mapping_hash(
|
|
item.governance.visibility_policy
|
|
),
|
|
},
|
|
)
|
|
session.commit()
|
|
return _datasource_response(item)
|
|
|
|
|
|
@router.get(
|
|
"/{datasource_id}/preview",
|
|
response_model=DatasourcePreviewResponse,
|
|
)
|
|
def api_preview_datasource(
|
|
datasource_id: str,
|
|
limit: int = Query(default=100, ge=1, le=500),
|
|
offset: int = Query(default=0, ge=0),
|
|
consistency: str = Query(default="current", pattern="^(current|live|frozen)$"),
|
|
materialization_ref: str | None = Query(default=None, max_length=120),
|
|
max_bytes: int = Query(default=1_000_000, ge=1_024, le=20_000_000),
|
|
timeout_ms: int = Query(default=2_000, ge=100, le=10_000),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourcePreviewResponse:
|
|
_require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
result = _provider().read_datasource(
|
|
session,
|
|
principal,
|
|
request=DatasourceReadRequest(
|
|
datasource_ref=f"datasource:{datasource_id}",
|
|
materialization_ref=materialization_ref,
|
|
consistency=consistency, # type: ignore[arg-type]
|
|
limit=limit,
|
|
offset=offset,
|
|
max_bytes=max_bytes,
|
|
timeout_ms=timeout_ms,
|
|
),
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourcePreviewResponse(
|
|
datasource=_datasource_response(result.datasource),
|
|
rows=[dict(row) for row in result.rows],
|
|
total_rows=result.total_rows,
|
|
truncated=result.truncated,
|
|
materialization=(
|
|
_materialization_response(result.materialization)
|
|
if result.materialization
|
|
else None
|
|
),
|
|
returned_bytes=result.returned_bytes,
|
|
elapsed_ms=result.elapsed_ms,
|
|
effective_row_limit=result.effective_row_limit,
|
|
effective_byte_limit=result.effective_byte_limit,
|
|
effective_timeout_ms=result.effective_timeout_ms,
|
|
diagnostics=[
|
|
DatasourcePreviewDiagnosticResponse(
|
|
severity=item.severity,
|
|
code=item.code,
|
|
message=item.message,
|
|
details=dict(item.details),
|
|
)
|
|
for item in result.diagnostics
|
|
],
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{datasource_id}/materializations",
|
|
response_model=DatasourceMaterializationListResponse,
|
|
)
|
|
def api_list_materializations(
|
|
datasource_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceMaterializationListResponse:
|
|
_require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
rows = _provider().list_materializations(
|
|
session,
|
|
principal,
|
|
datasource_ref=f"datasource:{datasource_id}",
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
return DatasourceMaterializationListResponse(
|
|
materializations=[_materialization_response(item) for item in rows]
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/{datasource_id}/refresh",
|
|
response_model=DatasourceStagePromoteResponse,
|
|
)
|
|
def api_refresh_datasource(
|
|
datasource_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceStagePromoteResponse:
|
|
_require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
try:
|
|
datasource, materialization = _provider().refresh_datasource(
|
|
session,
|
|
principal,
|
|
datasource_ref=f"datasource:{datasource_id}",
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.refreshed",
|
|
object_type="datasource",
|
|
object_id=datasource.ref,
|
|
details={
|
|
"materialization_ref": materialization.ref,
|
|
"revision": materialization.revision,
|
|
"fingerprint": materialization.fingerprint,
|
|
},
|
|
)
|
|
session.commit()
|
|
return DatasourceStagePromoteResponse(
|
|
datasource=_datasource_response(datasource),
|
|
materialization=_materialization_response(materialization),
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/{datasource_id}/refresh/stage",
|
|
response_model=DatasourceStageResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def api_prepare_refresh(
|
|
datasource_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceStageResponse:
|
|
_require_any_scope(
|
|
principal,
|
|
SOURCE_WRITE_SCOPE,
|
|
STAGE_WRITE_SCOPE,
|
|
ADMIN_SCOPE,
|
|
)
|
|
try:
|
|
stage = _provider().prepare_refresh(
|
|
session,
|
|
principal,
|
|
datasource_ref=f"datasource:{datasource_id}",
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.refresh.staged",
|
|
object_type="datasource_stage",
|
|
object_id=stage.ref,
|
|
details={
|
|
"datasource_ref": stage.target_datasource_ref,
|
|
"approval_state": stage.approval.get("state"),
|
|
"quality_policy_hash": stage.validation.get("policy_hash"),
|
|
},
|
|
)
|
|
session.commit()
|
|
return _stage_response(stage)
|
|
|
|
|
|
@router.post(
|
|
"/{datasource_id}/freeze",
|
|
response_model=DatasourceMaterializationResponse,
|
|
)
|
|
def api_freeze_datasource(
|
|
datasource_id: str,
|
|
payload: DatasourceFreezeRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceMaterializationResponse:
|
|
_require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
datasource_ref = f"datasource:{datasource_id}"
|
|
try:
|
|
materialization = _provider().freeze_datasource(
|
|
session,
|
|
principal,
|
|
datasource_ref=datasource_ref,
|
|
label=payload.label,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.frozen",
|
|
object_type="datasource",
|
|
object_id=datasource_ref,
|
|
details={
|
|
"materialization_ref": materialization.ref,
|
|
"revision": materialization.revision,
|
|
"label": materialization.frozen_label,
|
|
},
|
|
)
|
|
session.commit()
|
|
return _materialization_response(materialization)
|
|
|
|
|
|
@router.delete(
|
|
"/{datasource_id}",
|
|
response_model=DatasourceRetireResponse,
|
|
)
|
|
def api_retire_datasource(
|
|
datasource_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> DatasourceRetireResponse:
|
|
_require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE)
|
|
datasource_ref = f"datasource:{datasource_id}"
|
|
try:
|
|
_provider().retire_datasource(
|
|
session,
|
|
principal,
|
|
datasource_ref=datasource_ref,
|
|
)
|
|
except DatasourceError as exc:
|
|
raise _http_error(exc) from exc
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="datasources.retired",
|
|
object_type="datasource",
|
|
object_id=datasource_ref,
|
|
details={},
|
|
)
|
|
session.commit()
|
|
return DatasourceRetireResponse(retired=True, datasource_ref=datasource_ref)
|
|
|
|
|
|
def _datasource_response(item: DatasourceDescriptor) -> DatasourceResponse:
|
|
return DatasourceResponse(
|
|
ref=item.ref,
|
|
source_name=item.source_name,
|
|
name=item.name,
|
|
description=item.description,
|
|
kind=item.kind,
|
|
mode=item.mode,
|
|
shape=item.shape,
|
|
status=item.status,
|
|
provider=item.provider,
|
|
provider_ref=item.provider_ref,
|
|
schema=[
|
|
DatasourceFieldResponse(
|
|
name=field.name,
|
|
data_type=field.data_type,
|
|
nullable=field.nullable,
|
|
classification=field.classification,
|
|
)
|
|
for field in item.schema
|
|
],
|
|
schema_version=item.schema_version,
|
|
fingerprint=item.fingerprint,
|
|
current_materialization_ref=item.current_materialization_ref,
|
|
row_count=item.row_count,
|
|
byte_count=item.byte_count,
|
|
updated_at=item.updated_at.isoformat() if item.updated_at else None,
|
|
capabilities=list(item.capabilities),
|
|
provenance=dict(item.provenance),
|
|
metadata=dict(item.metadata),
|
|
governance=item.governance.to_dict(),
|
|
)
|
|
|
|
|
|
def _materialization_response(
|
|
item: DatasourceMaterialization,
|
|
) -> DatasourceMaterializationResponse:
|
|
return DatasourceMaterializationResponse(
|
|
ref=item.ref,
|
|
datasource_ref=item.datasource_ref,
|
|
revision=item.revision,
|
|
state=item.state,
|
|
fingerprint=item.fingerprint,
|
|
schema=[
|
|
DatasourceFieldResponse(
|
|
name=field.name,
|
|
data_type=field.data_type,
|
|
nullable=field.nullable,
|
|
classification=field.classification,
|
|
)
|
|
for field in item.schema
|
|
],
|
|
row_count=item.row_count,
|
|
byte_count=item.byte_count,
|
|
frozen_at=item.frozen_at.isoformat() if item.frozen_at else None,
|
|
frozen_label=item.frozen_label,
|
|
source_timestamp=(
|
|
item.source_timestamp.isoformat() if item.source_timestamp else None
|
|
),
|
|
created_at=item.created_at.isoformat() if item.created_at else None,
|
|
disposed_at=item.disposed_at.isoformat() if item.disposed_at else None,
|
|
disposition=dict(item.disposition),
|
|
provenance=dict(item.provenance),
|
|
metadata=dict(item.metadata),
|
|
governance=item.governance.to_dict(),
|
|
)
|
|
|
|
|
|
def _stage_response(item: DatasourceStage) -> DatasourceStageResponse:
|
|
return DatasourceStageResponse(
|
|
ref=item.ref,
|
|
name=item.name,
|
|
source_name=item.source_name,
|
|
kind=item.kind,
|
|
mode=item.mode,
|
|
shape=item.shape,
|
|
state=item.state,
|
|
target_datasource_ref=item.target_datasource_ref,
|
|
fingerprint=item.fingerprint,
|
|
schema=[
|
|
DatasourceFieldResponse(
|
|
name=field.name,
|
|
data_type=field.data_type,
|
|
nullable=field.nullable,
|
|
classification=field.classification,
|
|
)
|
|
for field in item.schema
|
|
],
|
|
row_count=item.row_count,
|
|
byte_count=item.byte_count,
|
|
validation=dict(item.validation),
|
|
approval=dict(item.approval),
|
|
created_at=item.created_at.isoformat() if item.created_at else None,
|
|
promoted_at=item.promoted_at.isoformat() if item.promoted_at else None,
|
|
promoted_materialization_ref=item.promoted_materialization_ref,
|
|
provenance=dict(item.provenance),
|
|
metadata=dict(item.metadata),
|
|
governance=item.governance.to_dict(),
|
|
)
|
|
|
|
|
|
def _origin_response(item: DatasourceOrigin) -> DatasourceOriginResponse:
|
|
return DatasourceOriginResponse(
|
|
ref=item.ref,
|
|
source_name=item.source_name,
|
|
name=item.name,
|
|
description=item.description,
|
|
kind=item.kind,
|
|
shape=item.shape,
|
|
supported_modes=list(item.supported_modes),
|
|
provider=item.provider,
|
|
schema=[
|
|
DatasourceFieldResponse(
|
|
name=field.name,
|
|
data_type=field.data_type,
|
|
nullable=field.nullable,
|
|
classification=field.classification,
|
|
)
|
|
for field in item.schema
|
|
],
|
|
schema_version=item.schema_version,
|
|
fingerprint=item.fingerprint,
|
|
row_count=item.row_count,
|
|
byte_count=item.byte_count,
|
|
updated_at=item.updated_at.isoformat() if item.updated_at else None,
|
|
capabilities=list(item.capabilities),
|
|
metadata=dict(item.metadata),
|
|
source_mode=item.source_mode,
|
|
pushdown=DatasourceOriginPushdownResponse(
|
|
projections=item.pushdown.projections,
|
|
pagination=item.pushdown.pagination,
|
|
filters=list(item.pushdown.filters),
|
|
aggregations=list(item.pushdown.aggregations),
|
|
sorting=list(item.pushdown.sorting),
|
|
),
|
|
health=DatasourceOriginHealthResponse(
|
|
status=item.health.status,
|
|
code=item.health.code,
|
|
summary=item.health.summary,
|
|
checked_at=(
|
|
item.health.checked_at.isoformat() if item.health.checked_at else None
|
|
),
|
|
details=dict(item.health.details),
|
|
),
|
|
)
|
|
|
|
|
|
def _evidence_response(item) -> DatasourceLifecycleEvidenceResponse:
|
|
return DatasourceLifecycleEvidenceResponse(
|
|
ref=f"lifecycle-evidence:{item.id}",
|
|
subject_ref=item.subject_ref,
|
|
event_type=item.event_type,
|
|
occurred_at=item.occurred_at.isoformat(),
|
|
actor_ref=item.actor_ref,
|
|
policy_version=item.policy_version,
|
|
policy_hash=item.policy_hash,
|
|
subject_digest=item.subject_digest,
|
|
previous_event_hash=item.previous_event_hash,
|
|
event_hash=item.event_hash,
|
|
details=dict(item.details_),
|
|
)
|
|
|
|
|
|
def _governance(
|
|
payload: DatasourceGovernancePayload | None,
|
|
) -> DatasourceGovernance | None:
|
|
if payload is None:
|
|
return None
|
|
return DatasourceGovernance.from_mapping(payload.model_dump())
|
|
|
|
|
|
def _safe_mapping(value: object) -> Mapping[str, object]:
|
|
return value if isinstance(value, Mapping) else {}
|
|
|
|
|
|
def _mapping_hash(value: Mapping[str, object]) -> str | None:
|
|
if not value:
|
|
return None
|
|
encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _audit(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
action: str,
|
|
object_type: str,
|
|
object_id: str,
|
|
details: dict[str, object],
|
|
) -> None:
|
|
audit_event(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
user_id=getattr(principal.user, "id", None),
|
|
api_key_id=principal.api_key_id,
|
|
action=action,
|
|
object_type=object_type,
|
|
object_id=object_id,
|
|
details=details,
|
|
)
|
|
|
|
|
|
__all__ = ["router"]
|