DataOps-platform/deployment/app/api/system/metering_showback_routes.py

"""Closed, Showback-only WP12 HTTP boundary; no chargeback endpoint exists."""

from __future__ import annotations

from flask import g, jsonify, request
from sqlalchemy.exc import SQLAlchemyError

from app.api.system import bp
from app.core.governance.metering_showback import DatabaseMeteringShowbackService
from app.core.system.permissions import (
    METERING_MANAGE,
    METERING_READ,
    require_permissions,
)
from app.models.result import failed, success

_MAX_BODY_BYTES = 4096
_EVENT_FIELDS = {"schema_version", "event_uid", "event_kind", "occurred_at", "window_start", "window_end", "quantity", "unit", "idempotency_key", "evidence", "mapping", "correction_of"}
_REQUIRED_EVENT_FIELDS = _EVENT_FIELDS - {"correction_of"}
_ALLOCATION_FIELDS = {"schema_version", "rule_uid", "rule_version", "effective_start", "effective_end", "mapping", "allocations"}
_BUDGET_FIELDS = {"schema_version", "budget_uid", "window", "mapping", "limit_micros", "threshold_micros"}


def _service() -> DatabaseMeteringShowbackService:
    return DatabaseMeteringShowbackService()


def _response(value: object, status: int = 200):
    response = jsonify(success(value, code=status))
    response.headers["Cache-Control"] = "no-store"
    return response, status


def _failure(status: int = 400):
    response = jsonify(failed("metering request rejected", code=status))
    response.headers["Cache-Control"] = "no-store"
    return response, status


def _event_body() -> dict:
    if request.content_length is not None and request.content_length > _MAX_BODY_BYTES:
        raise ValueError("metering_body_closed")
    value = request.get_json(silent=True)
    if not isinstance(value, dict) or set(value) not in (_REQUIRED_EVENT_FIELDS, _EVENT_FIELDS):
        raise ValueError("metering_body_closed")
    mapping = value.get("mapping")
    if not isinstance(mapping, dict) or set(mapping) != {"department", "project", "cost_center"}:
        raise ValueError("metering_mapping_closed")
    return value


@bp.post("/metering/events")
@require_permissions(METERING_MANAGE)
def record_metering_event():
    """record local metering event"""
    try:
        body = _event_body()
        result = _service().record_for_principal(principal_id=str(g.current_user["id"]), body=body)
        return _response(result, 201 if not result.get("replay") else 200)
    except (KeyError, LookupError, PermissionError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


def _control_body(fields: set[str]) -> dict:
    if request.content_length is not None and request.content_length > _MAX_BODY_BYTES:
        raise ValueError("metering_body_closed")
    value = request.get_json(silent=True)
    if not isinstance(value, dict) or set(value) != fields:
        raise ValueError("metering_body_closed")
    mapping = value.get("mapping")
    if not isinstance(mapping, dict) or set(mapping) != {"department", "project", "cost_center"}:
        raise ValueError("metering_mapping_closed")
    return value


@bp.post("/metering/allocations")
@require_permissions(METERING_MANAGE)
def publish_metering_allocation():
    """publish versioned local allocation rule"""
    try:
        return _response(_service().publish_allocation_for_principal(principal_id=str(g.current_user["id"]), body=_control_body(_ALLOCATION_FIELDS)), 201)
    except (KeyError, LookupError, PermissionError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


@bp.post("/metering/budgets")
@require_permissions(METERING_MANAGE)
def evaluate_metering_budget():
    """persist disabled-provider budget alert evidence"""
    try:
        return _response(_service().evaluate_budget_for_principal(principal_id=str(g.current_user["id"]), body=_control_body(_BUDGET_FIELDS)), 201)
    except (KeyError, LookupError, PermissionError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure()


def _window() -> str:
    if set(request.args) != {"window"}:
        raise ValueError("metering_query_closed")
    value = request.args.get("window", "")
    if len(value) != 7 or value[4:5] != "-" or not value.replace("-", "").isdigit():
        raise ValueError("metering_window_invalid")
    return value


@bp.get("/metering/showback")
@require_permissions(METERING_READ)
def get_showback():
    """get local showback"""
    try:
        return _response(_service().showback_for_principal(principal_id=str(g.current_user["id"]), window=_window(), kind="showback"))
    except (KeyError, LookupError, PermissionError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure(403)


@bp.get("/metering/reconciliation")
@require_permissions(METERING_READ)
def get_metering_reconciliation():
    """get metering reconciliation"""
    try:
        return _response(_service().showback_for_principal(principal_id=str(g.current_user["id"]), window=_window(), kind="reconciliation"))
    except (KeyError, LookupError, PermissionError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure(403)


@bp.get("/metering/allocation-replay")
@require_permissions(METERING_READ)
def get_metering_allocation_replay():
    """replay one persisted allocation-rule version"""
    try:
        if set(request.args) != {"window", "rule_uid", "rule_version"}:
            raise ValueError("metering_query_closed")
        version = request.args.get("rule_version", "")
        if not version.isdigit():
            raise ValueError("rule_version_invalid")
        return _response(_service().allocation_replay_for_principal(principal_id=str(g.current_user["id"]), window=request.args.get("window", ""), rule_uid=request.args.get("rule_uid", ""), rule_version=int(version)))
    except (KeyError, LookupError, PermissionError, RuntimeError, TypeError, ValueError, SQLAlchemyError):
        return _failure(403)