import uvicorn from fastapi import FastAPI, HTTPException, Query from fastapi.openapi.docs import get_swagger_ui_html from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse from pydantic import BaseModel from typing import Literal from datetime import date # ── Models ──────────────────────────────────────────────────────────────────── class Transaction(BaseModel): transaction_id: str date: date account: str amount: float status: Literal["Matched", "Unmatched", "Pending"] flag: Literal["None", "Duplicate", "Threshold Breach", "Manual Review"] class ReconSummary(BaseModel): total: int matched: int unmatched: int pending: int flagged: int # ── Sample data ─────────────────────────────────────────────────────────────── TRANSACTIONS: list[Transaction] = [ Transaction(transaction_id="TXN-001", date=date(2026, 4, 10), account="ACC-9821", amount=12400.00, status="Matched", flag="None"), Transaction(transaction_id="TXN-002", date=date(2026, 4, 11), account="ACC-4473", amount=3750.50, status="Unmatched", flag="Duplicate"), Transaction(transaction_id="TXN-003", date=date(2026, 4, 11), account="ACC-1190", amount=88200.00, status="Unmatched", flag="Threshold Breach"), Transaction(transaction_id="TXN-004", date=date(2026, 4, 12), account="ACC-6654", amount=540.00, status="Matched", flag="None"), Transaction(transaction_id="TXN-005", date=date(2026, 4, 13), account="ACC-3312", amount=21000.00, status="Pending", flag="Manual Review"), ] # ── App ─────────────────────────────────────────────────────────────────────── app = FastAPI( title="Recon Ranger", description="Financial crime reconciliation API — patrol your data landscape for inconsistencies.", version="0.1.0", docs_url=None, ) app.mount("/static", StaticFiles(directory="static"), name="static") # ── Endpoints ───────────────────────────────────────────────────────────────── @app.get( "/api/transactions", response_model=list[Transaction], summary="List transactions", tags=["Transactions"], ) async def list_transactions( status: Literal["Matched", "Unmatched", "Pending"] | None = Query(None, description="Filter by status"), flag: Literal["None", "Duplicate", "Threshold Breach", "Manual Review"] | None = Query(None, description="Filter by flag"), ) -> list[Transaction]: """Return all transactions, optionally filtered by status and/or flag.""" results = TRANSACTIONS if status: results = [t for t in results if t.status == status] if flag: results = [t for t in results if t.flag == flag] return results @app.get( "/api/transactions/{transaction_id}", response_model=Transaction, summary="Get a transaction", tags=["Transactions"], ) async def get_transaction(transaction_id: str) -> Transaction: """Retrieve a single transaction by ID.""" for t in TRANSACTIONS: if t.transaction_id == transaction_id: return t raise HTTPException(status_code=404, detail=f"{transaction_id} not found") @app.post( "/api/transactions", response_model=Transaction, status_code=201, summary="Submit a transaction", tags=["Transactions"], ) async def create_transaction(transaction: Transaction) -> Transaction: """Submit a new transaction for reconciliation.""" for t in TRANSACTIONS: if t.transaction_id == transaction.transaction_id: raise HTTPException(status_code=409, detail=f"{transaction.transaction_id} already exists") TRANSACTIONS.append(transaction) return transaction @app.get( "/api/summary", response_model=ReconSummary, summary="Reconciliation summary", tags=["Reporting"], ) async def get_summary() -> ReconSummary: """Return aggregate counts across all transactions.""" return ReconSummary( total=len(TRANSACTIONS), matched=sum(1 for t in TRANSACTIONS if t.status == "Matched"), unmatched=sum(1 for t in TRANSACTIONS if t.status == "Unmatched"), pending=sum(1 for t in TRANSACTIONS if t.status == "Pending"), flagged=sum(1 for t in TRANSACTIONS if t.flag != "None"), ) @app.delete( "/api/transactions/{transaction_id}", status_code=204, summary="Delete a transaction", tags=["Transactions"], ) async def delete_transaction(transaction_id: str) -> None: """Remove a transaction by ID.""" for i, t in enumerate(TRANSACTIONS): if t.transaction_id == transaction_id: TRANSACTIONS.pop(i) return raise HTTPException(status_code=404, detail=f"{transaction_id} not found") @app.get("/api/docs", include_in_schema=False) async def swagger_ui_dark() -> HTMLResponse: base = get_swagger_ui_html( openapi_url=app.openapi_url, title="Recon Ranger — API Docs", swagger_css_url="/static/api/swagger-ui.css", ) html = base.body.decode() # Inject dark theme + main site styles html = html.replace( "", ( '\n' '\n' "\n" "" ), ) # Inject the site nav bar before Swagger's root div nav_html = """
""" html = html.replace("", "" + nav_html) html = html.replace("", "
") return HTMLResponse(html) if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)