DataOps-platform/app/api/meta_data/active_metadata.py

"""HTTP API for active metadata discovery and field lineage."""

from __future__ import annotations

from flask import g, jsonify, request

from app import db
from app.api.meta_data import bp
from app.core.meta_data.active_metadata import (
    ActiveMetadataConflict,
    ActiveMetadataError,
    ActiveMetadataNotFound,
    ActiveMetadataService,
)
from app.core.meta_data.active_metadata_repository import (
    SqlAlchemyActiveMetadataRepository,
)
from app.models.result import failed, success


def _service():
    return ActiveMetadataService(SqlAlchemyActiveMetadataRepository(db.session))


def _error(exc):
    db.session.rollback()
    if isinstance(exc, ActiveMetadataNotFound):
        return jsonify(failed(str(exc), code=404)), 404
    if isinstance(exc, ActiveMetadataConflict):
        return jsonify(failed(str(exc), code=409)), 409
    if isinstance(exc, ActiveMetadataError):
        return jsonify(failed(str(exc), code=400)), 400
    raise exc


@bp.route("/active-metadata/plans", methods=["GET"])
def list_active_metadata_plans():
    return jsonify(success(_service().list_plans()))


@bp.route("/active-metadata/plans", methods=["POST"])
def create_active_metadata_plan():
    try:
        result = _service().create_plan(
            request.get_json(silent=True),
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "主动发现计划已创建")), 201
    except Exception as exc:
        return _error(exc)

@bp.route("/active-metadata/plans/<plan_uid>/runs", methods=["POST"])
def execute_active_metadata_plan(plan_uid):
    batch_key = str(request.headers.get("Idempotency-Key") or "").strip()
    if not batch_key:
        return jsonify(failed("Idempotency-Key is required", code=428)), 428
    try:
        result = _service().execute(
            plan_uid,
            request.get_json(silent=True),
            batch_key=batch_key,
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "主动发现批次已完成"))
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/plans/<plan_uid>/runs", methods=["GET"])
def list_active_metadata_runs(plan_uid):
    try:
        return jsonify(success(_service().list_runs(plan_uid)))
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/assets", methods=["GET"])
def list_active_metadata_assets():
    try:
        return jsonify(
            success(_service().list_assets(request.args.get("source_uid")))
        )
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/runs/<run_uid>/changes", methods=["GET"])
def list_active_metadata_changes(run_uid):
    try:
        return jsonify(success(_service().list_changes(run_uid)))
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/runs/<run_uid>/lineage", methods=["GET"])
def list_active_metadata_lineage(run_uid):
    try:
        return jsonify(success(_service().list_lineage(run_uid)))
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/runs/<run_uid>/health-signals", methods=["GET"])
def list_active_metadata_health(run_uid):
    try:
        return jsonify(success(_service().list_health_signals(run_uid)))
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/corrections", methods=["GET"])
def list_active_metadata_corrections():
    try:
        return jsonify(
            success(_service().list_corrections(request.args.get("asset_uid")))
        )
    except Exception as exc:
        return _error(exc)


@bp.route("/active-metadata/assets/<asset_uid>/corrections", methods=["POST"])
def submit_active_metadata_correction(asset_uid):
    try:
        result = _service().submit_correction(
            asset_uid,
            request.get_json(silent=True),
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "元数据纠错已提交")), 201
    except Exception as exc:
        return _error(exc)


@bp.route(
    "/active-metadata/corrections/<correction_uid>/resolve",
    methods=["POST"],
)
def resolve_active_metadata_correction(correction_uid):
    body = request.get_json(silent=True) or {}
    try:
        result = _service().resolve_correction(
            correction_uid,
            expected_version=body.get("expected_version"),
            decision=body.get("decision"),
            resolution=body.get("resolution") or {},
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "元数据纠错已处置"))
    except Exception as exc:
        return _error(exc)