LedgerLine/services/sanctions/app.py

import asyncio

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from opentelemetry import trace
from opentelemetry.trace import StatusCode
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

from common.db import engine
from common.logging_config import configure_logging
from common.telemetry import setup_telemetry

from . import chaos
from .repository import is_match

SERVICE_NAME = "sanctions"
logger = configure_logging(SERVICE_NAME)
tracer = trace.get_tracer(__name__)
setup_telemetry(service_name=SERVICE_NAME)
app = FastAPI(title=SERVICE_NAME)

FastAPIInstrumentor.instrument_app(
    app,
    excluded_urls="/healthz",
    exclude_spans=["receive", "send"],
)
SQLAlchemyInstrumentor().instrument(engines=[engine.sync_engine])

class LookupRequest(BaseModel):
    holder_name: str
    request_id: str = "unknown"


@app.get("/")
async def root():
    return {"service": SERVICE_NAME, "version": "0.1.0"}


@app.get("/healthz")
async def healthz(request: Request):
    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
        logger.info(f"received chatty-health healthz probe from gateway")
    return {"status": "ok", "service": SERVICE_NAME}


@app.get("/readyz")
async def readyz():
    try:
        async with engine.connect() as conn:
            await conn.exec_driver_sql("SELECT 1")
    except Exception as exc:
        span = trace.get_current_span()
        span.record_exception(exc)
        span.set_status(StatusCode.ERROR, str(exc))
        logger.exception(f"readyz check failed: {exc}")
        return JSONResponse(status_code=503, content={"status": "unavailable"})
    return {"status": "ok"}


@app.post("/internal/lookup")
async def lookup(body: LookupRequest):
    logger.info(f"lookup for holder_name={body.holder_name!r} request_id={body.request_id}")

    if await chaos.sanctions_errors_enabled() and chaos.should_error():
        logger.info(f"sanctions_errors chaos triggered for request_id={body.request_id}")
        return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"})

    if await chaos.slow_sanctions_enabled():
        delay = chaos.sample_delay_seconds()
        logger.info(f"slow_sanctions chaos sleeping {delay:.3f}s for request_id={body.request_id}")
        await asyncio.sleep(delay)

    # holder_name is PII: it stays off the span; only the outcome is recorded.
    with tracer.start_as_current_span("sanctions.lookup_watchlist") as span:
        match = await is_match(body.holder_name)
        span.set_attribute("ledgerline.sanctions.matched", match is not None)
        if match:
            span.set_attribute("ledgerline.sanctions.list_source", match["list_source"])
    if match:
        return {"matched": True, **match}
    return {"matched": False, "country": None, "list_source": None}