DataOps-platform/deployment/app/api/data_rules/quality_routes.py

"""HTTP boundary for generic, deterministic quality operations."""

from __future__ import annotations

from flask import g, jsonify, request

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


def get_quality_operations_service():
    from app.core.data_rules.quality_operations import QualityOperationsService
    from app.core.data_rules.quality_repository import (
        SqlAlchemyQualityOperationsRepository,
    )

    return QualityOperationsService(
        SqlAlchemyQualityOperationsRepository(db.session),
        publish_authorizer=lambda _actor_uid: None,
        commit=db.session.commit,
        rollback=db.session.rollback,
    )


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


def _payload():
    value = request.get_json(silent=True)
    if not isinstance(value, dict):
        raise ValueError("request body must be an object")
    return value


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.get("/quality-operations/templates")
def list_quality_templates():
    try:
        return jsonify(
            success(get_quality_operations_service().list_templates())
        ), 200
    except Exception as error:
        return _error(error)


@bp.post("/quality-operations/templates")
def create_quality_template():
    try:
        record = get_quality_operations_service().create_template(
            _payload(),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.post("/quality-operations/templates/<template_uid>/revisions")
def revise_quality_template(template_uid):
    try:
        payload = _payload()
        record = get_quality_operations_service().revise_template(
            template_uid,
            payload.get("definition"),
            expected_version=int(payload.get("expected_version")),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.post("/quality-operations/templates/<template_uid>/publish")
def publish_quality_template(template_uid):
    try:
        payload = _payload()
        record = get_quality_operations_service().publish_template(
            template_uid,
            expected_version=int(payload.get("expected_version")),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.post("/quality-operations/execute")
def execute_quality_profile():
    try:
        record = get_quality_operations_service().execute(
            _payload(),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.get("/quality-operations/runs")
def list_quality_runs():
    try:
        records = get_quality_operations_service().list_runs(
            request.args.get("asset_uid")
        )
        return jsonify(success(records)), 200
    except Exception as error:
        return _error(error)


@bp.get("/quality-operations/runs/<run_uid>")
def get_quality_run(run_uid):
    try:
        return jsonify(
            success(get_quality_operations_service().get_run(run_uid))
        ), 200
    except Exception as error:
        return _error(error)


@bp.get("/quality-operations/assets/<asset_uid>/trend")
def get_quality_trend(asset_uid):
    try:
        return jsonify(
            success(get_quality_operations_service().trend(asset_uid))
        ), 200
    except Exception as error:
        return _error(error)