Business Event Logging for FastAPI

pip install httpx

# app/logit.py
import os
import httpx
from typing import Optional

LOGIT_ENDPOINT = "https://api.logit.now/api/ingest"
_client: Optional[httpx.AsyncClient] = None


def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None or _client.is_closed:
        _client = httpx.AsyncClient(timeout=5.0)
    return _client


async def log_event(
    channel: str,
    event: str,
    description: str,
    icon: str = "📌",
    notify: bool = False,
    tags: Optional[dict] = None,
    metadata: Optional[dict] = None,
) -> None:
    try:
        await get_client().post(
            LOGIT_ENDPOINT,
            headers={"Authorization": f"Bearer {os.environ['LOGIT_API_KEY']}"},
            json={
                "channel": channel,
                "event": event,
                "description": description,
                "icon": icon,
                "notify": notify,
                "tags": tags or {},
                "metadata": metadata or {},
            },
        )
    except Exception as exc:
        print(f"logit: {exc}")

# app/routers/auth.py
from fastapi import APIRouter
from app.logit import log_event

router = APIRouter()

@router.post("/signup")
async def signup(email: str, plan: str):
    user = await create_user(email, plan)

    await log_event(
        channel="signups",
        event="New user signed up",
        description=user.email,
        icon="👤",
        notify=True,
        tags={"plan": plan, "source": "api"},
        metadata={"userId": str(user.id), "email": user.email},
    )

    return {"id": str(user.id)}

from fastapi import BackgroundTasks

@router.post("/payment-webhook")
async def handle_payment(payload: dict, background_tasks: BackgroundTasks):
    await process_payment(payload)

    background_tasks.add_task(
        log_event,
        channel="payments",
        event="Payment received",
        description=f"{payload.get('customer_email')} — {payload.get('amount')}",
        icon="💰",
        notify=True,
        tags={"currency": payload.get("currency", "usd")},
        metadata=payload,
    )

    return {"ok": True}