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.metrics import observe_db_pools
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])
observe_db_pools({"default": engine})

class LookupRequest(BaseModel):
    holder_name: str


@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("healthz_probe_received", probe="chatty-health")
    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("readyz_check_failed")
        return JSONResponse(status_code=503, content={"status": "unavailable"})
    return {"status": "ok"}


@app.post("/internal/lookup")
async def lookup(body: LookupRequest):
    # holder_name is PII: logged by omission, not by redaction.
    logger.info("lookup_received")

    if await chaos.sanctions_errors_enabled() and chaos.should_error():
        logger.info("chaos_engaged", chaos_flag="sanctions_errors")
        return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"})

    if await chaos.slow_sanctions_enabled():
        delay = chaos.sample_delay_seconds()
        logger.info("chaos_engaged", chaos_flag="slow_sanctions", delay_ms=round(delay * 1000, 1))
        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"])
    logger.info(
        "holder_screened",
        matched=match is not None,
        list_source=match["list_source"] if match else None,
    )
    if match:
        return {"matched": True, **match}
    return {"matched": False, "country": None, "list_source": None}