FastAPI with logging and proxying to target with blacklist support

fastapi_proxy_logging.py

import json
from datetime import datetime
from typing import List
from fastapi import FastAPI, Request, HTTPException
from starlette.responses import JSONResponse
import httpx

app = FastAPI()

BLACKLIST_PATHS: List[str] = [
    # Add any paths you want to blacklist here.
]

async def log_request(client_ip, current_datetime, input_payload, output_payload):
    log_data = {
        "client_ip": client_ip,
        "datetime": current_datetime.isoformat(),
        "input": input_payload,
        "output": output_payload,
    }
    log_json = json.dumps(log_data, indent=4)
    # Replace this with your own logging implementation.
    print(log_json)

async def fetch_proxy_request(url, request, body_data):
    async with httpx.AsyncClient() as client:
        try:
            response = await client.request(
                request.method,
                url,
                headers=request.headers,
                data=body_data,
            )
            return response
        except httpx.ConnectTimeout as exc:
            error_message = f"Connection timeout: {exc}"
        except httpx.ReadTimeout as exc:
            error_message = f"Read timeout: {exc}"
        except httpx.ConnectError as exc:
            error_message = f"Connection error: {exc}"
        except httpx.RequestError as exc:
            error_message = f"Request error: {exc}"
        return JSONResponse(status_code=500, content={"detail": error_message})

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"])
async def catch_all(request: Request, path: str):
    if path in BLACKLIST_PATHS:
        raise HTTPException(status_code=403, detail="Access to this path is forbidden")

    target_url = f"https://target-api.example.com/{path}"
    body_data = await request.body()
    proxy_response = await fetch_proxy_request(target_url, request, body_data)

    client_ip = request.client.host
    current_datetime = datetime.utcnow()
    input_payload = {
        "path": path,
        "method": request.method,
        "headers": dict(request.headers),
        "body": body_data.decode("utf-8"),
    }
    output_payload = {
        "status_code": proxy_response.status_code,
        "headers": dict(proxy_response.headers),
        "body": proxy_response.content.decode("utf-8"),
    }

    await log_request(client_ip, current_datetime, input_payload, output_payload)

    return JSONResponse(
        status_code=proxy_response.status_code,
        content=json.loads(proxy_response.content) if isinstance(proxy_response, JSONResponse) else proxy_response.json(),
        headers=proxy_response.headers,
    )