DataOps-platform/deployment/app/api/data_factory/observability_routes.py

"""HTTP boundary for SLO monitoring and evidence-bound data incidents."""

from __future__ import annotations

from flask import g, jsonify, request

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


def get_data_observability_service():
    from app.core.events.data_observability import DataObservabilityService
    from app.core.events.data_observability_repository import (
        SqlAlchemyDataObservabilityRepository,
    )

    return DataObservabilityService(
        SqlAlchemyDataObservabilityRepository(db.session),
        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("/observability/overview")
def get_observability_overview():
    try:
        return jsonify(success(get_data_observability_service().overview())), 200
    except Exception as error:
        return _error(error)


@bp.get("/observability/slos")
def list_slo_policies():
    try:
        return jsonify(success(get_data_observability_service().list_slos())), 200
    except Exception as error:
        return _error(error)


@bp.post("/observability/slos")
def create_slo_policy():
    try:
        record = get_data_observability_service().create_slo(
            _payload(), actor_uid=_actor_uid()
        )
        return jsonify(success(record)), 201
    except Exception as error:
        return _error(error)


@bp.post("/observability/collect")
def collect_observability_signals():
    try:
        _payload()
        record = get_data_observability_service().collect(
            actor_uid=_actor_uid()
        )
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.get("/observability/alerts")
def list_observability_alerts():
    try:
        return jsonify(
            success(get_data_observability_service().list_alerts())
        ), 200
    except Exception as error:
        return _error(error)


@bp.post("/observability/alerts/<alert_uid>/suppress")
def suppress_observability_alert(alert_uid):
    try:
        payload = _payload()
        record = get_data_observability_service().suppress_alert(
            alert_uid,
            until=payload.get("until"),
            reason=payload.get("reason"),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.post("/observability/alerts/<alert_uid>/acknowledge-delivery")
def acknowledge_observability_alert_delivery(alert_uid):
    try:
        payload = _payload()
        record = get_data_observability_service().acknowledge_delivery(
            alert_uid,
            channel=payload.get("channel"),
            receipt_id=payload.get("receipt_id"),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.get("/observability/incidents")
def list_data_incidents():
    try:
        return jsonify(
            success(get_data_observability_service().list_incidents())
        ), 200
    except Exception as error:
        return _error(error)


@bp.get("/observability/incidents/<incident_uid>")
def get_data_incident(incident_uid):
    try:
        return jsonify(
            success(
                get_data_observability_service().incident_detail(incident_uid)
            )
        ), 200
    except Exception as error:
        return _error(error)


@bp.post("/observability/incidents/<incident_uid>/escalate")
def escalate_data_incident(incident_uid):
    try:
        payload = _payload()
        record = get_data_observability_service().escalate_incident(
            incident_uid,
            reason=payload.get("reason"),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)


@bp.post("/observability/incidents/<incident_uid>/close")
def close_data_incident(incident_uid):
    try:
        record = get_data_observability_service().close_incident(
            incident_uid,
            _payload(),
            actor_uid=_actor_uid(),
        )
        return jsonify(success(record)), 200
    except Exception as error:
        return _error(error)