DataOps-platform/deployment/app/api/data_development/semantic_routes.py

"""HTTP boundary for governed terms, code sets, metrics and field mappings."""

from __future__ import annotations

import io

from flask import g, jsonify, request, send_file

from app import db
from app.api.data_development import bp
from app.models.result import failed, success


def get_semantic_governance_service():
    from app.core.data_research.semantic_governance import (
        SemanticGovernanceService,
    )
    from app.core.data_research.semantic_repository import (
        SqlAlchemySemanticGovernanceRepository,
    )
    from app.core.events.outbox import enqueue_outbox

    return SemanticGovernanceService(
        SqlAlchemySemanticGovernanceRepository(db.session),
        outbox_enqueue=lambda **event: enqueue_outbox(db.session, **event),
    )


def _actor_uid():
    identity = getattr(g, "current_user", {}) or {}
    return identity.get("id") or identity.get("sub")


def _error(error):
    db.session.rollback()
    if isinstance(error, LookupError):
        return jsonify(failed(str(error), code=404)), 404
    if isinstance(error, ValueError):
        return jsonify(failed(str(error), code=400)), 400
    if isinstance(error, RuntimeError):
        return jsonify(failed(str(error), code=409)), 409
    return jsonify(failed("语义治理请求处理失败", code=500)), 500


@bp.route("/semantic-assets", methods=["GET"])
def list_semantic_assets():
    try:
        records = get_semantic_governance_service().list(
            asset_kind=request.args.get("asset_kind"),
            status=request.args.get("status"),
        )
        return jsonify(success(records)), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets", methods=["POST"])
def create_semantic_asset():
    payload = request.get_json(silent=True) or {}
    try:
        record = get_semantic_governance_service().create_draft(
            payload.get("asset_kind"),
            {key: value for key, value in payload.items() if key != "asset_kind"},
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>", methods=["GET"])
def get_semantic_asset(asset_uid):
    try:
        return jsonify(
            success(get_semantic_governance_service().get(asset_uid))
        ), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/versions", methods=["GET"])
def list_semantic_asset_versions(asset_uid):
    try:
        return jsonify(
            success(
                get_semantic_governance_service().list_versions(asset_uid)
            )
        ), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/reviews", methods=["GET"])
def list_semantic_asset_reviews(asset_uid):
    try:
        return jsonify(
            success(get_semantic_governance_service().list_reviews(asset_uid))
        ), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/audits", methods=["GET"])
def list_semantic_asset_audits(asset_uid):
    try:
        return jsonify(
            success(get_semantic_governance_service().list_audits(asset_uid))
        ), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/revisions", methods=["POST"])
def revise_semantic_asset(asset_uid):
    payload = request.get_json(silent=True) or {}
    try:
        record = get_semantic_governance_service().revise(
            asset_uid,
            payload.get("definition") or {},
            expected_version=int(payload.get("expected_version")),
            reason=payload.get("reason"),
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/submit", methods=["POST"])
def submit_semantic_asset(asset_uid):
    payload = request.get_json(silent=True) or {}
    try:
        record = get_semantic_governance_service().submit(
            asset_uid,
            expected_version=int(payload.get("expected_version")),
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/review", methods=["POST"])
def review_semantic_asset(asset_uid):
    payload = request.get_json(silent=True) or {}
    try:
        record = get_semantic_governance_service().review(
            asset_uid,
            expected_version=int(payload.get("expected_version")),
            decision=payload.get("decision"),
            reason=payload.get("reason"),
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/publish", methods=["POST"])
def publish_semantic_asset(asset_uid):
    payload = request.get_json(silent=True) or {}
    try:
        record = get_semantic_governance_service().publish(
            asset_uid,
            expected_version=int(payload.get("expected_version")),
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/<asset_uid>/rollback", methods=["POST"])
def rollback_semantic_asset(asset_uid):
    payload = request.get_json(silent=True) or {}
    try:
        record = get_semantic_governance_service().rollback(
            asset_uid,
            target_version=int(payload.get("target_version")),
            expected_version=int(payload.get("expected_version")),
            reason=payload.get("reason"),
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-field-mappings", methods=["GET"])
def list_semantic_field_mappings():
    try:
        records = get_semantic_governance_service().list_field_mappings(
            asset_uid=request.args.get("asset_uid"),
            data_element_uid=request.args.get("data_element_uid"),
        )
        return jsonify(success(records)), 200
    except Exception as error:
        return _error(error)


@bp.route("/semantic-field-mappings", methods=["POST"])
def create_semantic_field_mapping():
    try:
        record = get_semantic_governance_service().map_physical_field(
            request.get_json(silent=True) or {},
            actor_uid=_actor_uid(),
        )
        db.session.commit()
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.route("/semantic-assets/export", methods=["GET"])
def export_semantic_assets():
    from app.core.data_research.semantic_governance import SemanticExchange

    export_format = str(request.args.get("format") or "json").lower()
    try:
        assets = get_semantic_governance_service().list(status="published")
        exchange = SemanticExchange()
        if export_format == "rdf":
            content = exchange.export_rdf(assets)
            filename = "semantic-governance.rdf"
            mimetype = "application/rdf+xml"
        elif export_format == "json":
            content = exchange.export_json(assets)
            filename = "semantic-governance.json"
            mimetype = "application/json"
        else:
            raise ValueError("semantic export format is unsupported")
        return send_file(
            io.BytesIO(content),
            mimetype=mimetype,
            as_attachment=True,
            download_name=filename,
        )
    except Exception as error:
        return _error(error)