Blame for sinetti_api/errors.py

from __future__ import annotations

import uuid

from fastapi import Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

import obs
from sinetti_services import diagnostics_contract
from sinetti_services.tenant_context import TenantRequired


_ACTION_UNAVAILABLE_BODY = {
    "ok": False,
    "code": "action_unavailable",
    "message": "We couldn’t complete this action. Please try again later or contact support.",
}


def request_trace_id(request: Request) -> str:
    trace_id = str(getattr(request.state, "trace_id", "") or "")
    if trace_id:
        return trace_id
    trace_id = str(uuid.uuid4())
    request.state.trace_id = trace_id
    return trace_id


async def http_exception_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse:
    code = {
        400: "invalid_request",
        401: "unauthorized",
        403: "forbidden",
        404: "not_found",
        409: "conflict",
        422: "validation_error",
        503: "service_unavailable",
    }.get(exc.status_code, "http_error")
    trace_id = request_trace_id(request)
    headers = {**(exc.headers or {}), "X-Request-ID": trace_id}
    return JSONResponse(
        status_code=exc.status_code,
        headers=headers,
        content={
            "error": {
                "code": code,
                "message": str(exc.detail or "Request failed"),
                "trace_id": trace_id,
            }
        },
    )


async def request_validation_exception_handler(

    request: Request,

    exc: RequestValidationError,

) -> JSONResponse:
    trace_id = request_trace_id(request)
    # Locations and error types only — never the rejected input, which carries user content.
    try:
        fields = ",".join(
            sorted(
                {
                    ".".join(str(part) for part in (error.get("loc") or ())[:4])
                    + ":"
                    + str(error.get("type") or "")
                    for error in (exc.errors() or ())
                }
            )
        )[:400]
    except Exception:  # noqa: BLE001 — diagnostics must never mask the 422
        fields = ""
    obs.error(
        "request_validation_failed",
        trace_id=trace_id,
        method=request.method,
        path=diagnostics_contract.log_path(request),
        fields=fields,
    )
    return JSONResponse(
        status_code=422,
        headers={"X-Request-ID": trace_id},
        content={
            "error": {
                "code": "validation_error",
                "message": "Request validation failed.",
                "trace_id": trace_id,
            }
        },
    )


async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    trace_id = request_trace_id(request)
    obs.error(
        "request_failed",
        trace_id=trace_id,
        method=request.method,
        path=diagnostics_contract.log_path(request),
        exception_class=type(exc).__name__,
    )
    # Best-effort ops ledger append — never fail the HTTP 500 path if the write fails.
    try:
        from sinetti_services import ops_error_ledger

        # Class name only — never str(exc); exception text can carry secrets.
        ops_error_ledger.record_error(
            source="server",
            trace_id=trace_id,
            message=type(exc).__name__,
            path=diagnostics_contract.log_path(request),
        )
    except Exception as ledger_exc:  # noqa: BLE001 — ledger must not mask the original 500
        obs.error(
            "ops_error_ledger_write_failed",
            trace_id=trace_id,
            error_type=type(ledger_exc).__name__,
            error=str(ledger_exc),
        )
    return JSONResponse(
        status_code=500,
        headers={"X-Request-ID": trace_id},
        content={
            "error": {
                "code": "internal_error",
                "message": "The request could not be completed.",
                "trace_id": trace_id,
            }
        },
    )


class ActionUnavailable(Exception):
    """A protected action could not safely be completed; its reason remains internal."""

    def __init__(self, internal_reason: str) -> None:
        self.internal_reason = internal_reason
        super().__init__(internal_reason)


async def action_unavailable_handler(request: Request, exc: ActionUnavailable) -> JSONResponse:
    trace_id = request_trace_id(request)
    obs.error(
        "action_unavailable",
        internal_reason=exc.internal_reason,
        trace_id=trace_id,
        method=request.method,
        path=diagnostics_contract.log_path(request),
    )
    return JSONResponse(status_code=429, content=_ACTION_UNAVAILABLE_BODY)


# Human-readable names for the plan-blocked message. Keep in sync with store_billing.FEATURE_KEYS.
_FEATURE_DISPLAY_NAMES = {
    "counterparty_dd": "Counterparty Due Diligence",
    "business_footprint": "Company research",
    "csv_export": "CSV Export",
    "pdf_deep_dives": "PDF Deep-Dives",
    "market_balances": "Market Balances",
    "batch_check": "Batch check",
    "research_data": "Research data",
    "plant_table": "Plant table",
}


class FeatureBlocked(Exception):
    """Raised by sinetti_api.deps.require_feature when the caller's plan lacks `feature`.



    Handled as a graceful HTTP 200 envelope (never a bare 403) — see feature_blocked_handler.

    """

    def __init__(self, feature: str, *, reason: str = "plan", message: str = "") -> None:
        self.feature = feature
        self.reason = reason
        self.message = message
        super().__init__(f"feature not entitled: {feature}")


async def feature_blocked_handler(request: Request, exc: FeatureBlocked) -> JSONResponse:
    name = _FEATURE_DISPLAY_NAMES.get(exc.feature, exc.feature)
    return JSONResponse(
        status_code=200,
        content={
            "ok": False,
            "blocked": True,
            "reason": exc.reason,
            "feature": exc.feature,
            "message": exc.message
            or f"{name} is not included in your plan — contact your administrator.",
        },
    )


async def tenant_required_handler(request: Request, exc: TenantRequired) -> JSONResponse:
    trace_id = request_trace_id(request)
    message = str(exc) or "Organisation access must be assigned before using this feature."
    return JSONResponse(
        status_code=403,
        headers={"X-Request-ID": trace_id},
        content={
            "error": {
                "code": "tenant_required",
                "message": message,
                "trace_id": trace_id,
            }
        },
    )