Example: Analytics Dashboard with FastAPI + CIUF¶

An analytics dashboard API built with FastAPI and CIUF. Three endpoints — each hits PostgreSQL once and returns from memory on every request after that:

Daily revenue per customer plan (for a line chart)
Top 10 products by revenue this month
Per-store order metrics

Schema¶

CREATE TABLE customers (
    id      SERIAL PRIMARY KEY,
    name    TEXT NOT NULL,
    plan    TEXT NOT NULL CHECK (plan IN ('free', 'pro', 'enterprise'))
);

CREATE TABLE products (
    id       SERIAL PRIMARY KEY,
    name     TEXT NOT NULL,
    category TEXT NOT NULL,
    price    NUMERIC(10,2) NOT NULL
);

CREATE TABLE orders (
    id          SERIAL PRIMARY KEY,
    customer_id INT REFERENCES customers(id),
    product_id  INT REFERENCES products(id),
    store_id    INT NOT NULL,
    amount      NUMERIC(10,2) NOT NULL,
    status      TEXT NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Setup¶

pip install ciuf[postgres] fastapi uvicorn

Full application¶

# app.py
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any

from fastapi import FastAPI, HTTPException
from ciuf import Engine, from_sql

# ── Engine ────────────────────────────────────────────────────────────────────

DB_URL = "postgresql://user:password@localhost:5432/analytics"

engine: Engine = None  # initialised in lifespan


@asynccontextmanager
async def lifespan(app: FastAPI):
    global engine
    engine = Engine(DB_URL, max_memory_mb=512, ttl_seconds=3600)
    yield
    engine.dispose()


app = FastAPI(title="Analytics API", lifespan=lifespan)

# ── Queries ───────────────────────────────────────────────────────────────────

DAILY_REVENUE_SQL = """
    SELECT
        date_trunc('day', orders.created_at)  AS day,
        customers.plan,
        SUM(orders.amount)                    AS revenue,
        COUNT(*)                              AS order_count
    FROM orders
    JOIN customers ON orders.customer_id = customers.id
    WHERE orders.status = 'completed'
    GROUP BY date_trunc('day', orders.created_at), customers.plan
    ORDER BY day DESC
    LIMIT 90
"""

TOP_PRODUCTS_SQL = """
    SELECT
        products.name,
        products.category,
        SUM(orders.amount)  AS revenue,
        COUNT(*)            AS units_sold
    FROM orders
    JOIN products ON orders.product_id = products.id
    WHERE orders.status = 'completed'
    GROUP BY products.name, products.category
    ORDER BY revenue DESC
    LIMIT 10
"""

STORE_METRICS_SQL = """
    SELECT
        orders.store_id,
        COUNT(*)            AS total_orders,
        SUM(orders.amount)  AS total_revenue,
        AVG(orders.amount)  AS avg_order_value,
        customers.plan
    FROM orders
    JOIN customers ON orders.customer_id = customers.id
    WHERE orders.status = 'completed'
    GROUP BY orders.store_id, customers.plan
    ORDER BY total_revenue DESC
"""

# Build DAG nodes once at startup — warm on first request per node
_daily_revenue  = None
_top_products   = None
_store_metrics  = None


def get_daily_revenue():
    global _daily_revenue
    if _daily_revenue is None:
        _daily_revenue = from_sql(engine, DAILY_REVENUE_SQL)
    return _daily_revenue


def get_top_products():
    global _top_products
    if _top_products is None:
        _top_products = from_sql(engine, TOP_PRODUCTS_SQL)
    return _top_products


def get_store_metrics():
    global _store_metrics
    if _store_metrics is None:
        _store_metrics = from_sql(engine, STORE_METRICS_SQL)
    return _store_metrics


# ── Endpoints ─────────────────────────────────────────────────────────────────

@app.get("/dashboard/revenue")
def daily_revenue() -> list[dict[str, Any]]:
    """Daily revenue grouped by customer plan. Returns from cache after first call."""
    df = get_daily_revenue().query()
    df["day"] = df["day"].dt.strftime("%Y-%m-%d")
    df["revenue"] = df["revenue"].astype(float)
    return df.to_dict(orient="records")


@app.get("/dashboard/top-products")
def top_products() -> list[dict[str, Any]]:
    """Top 10 products by revenue. Cached after first call."""
    df = get_top_products().query()
    df["revenue"] = df["revenue"].astype(float)
    return df.to_dict(orient="records")


@app.get("/dashboard/stores")
def store_metrics() -> list[dict[str, Any]]:
    """Per-store order metrics. Cached after first call."""
    df = get_store_metrics().query()
    df["total_revenue"]    = df["total_revenue"].astype(float)
    df["avg_order_value"]  = df["avg_order_value"].astype(float)
    return df.to_dict(orient="records")


# ── Write endpoints (keep cache in sync) ─────────────────────────────────────

@app.post("/orders", status_code=201)
def create_order(
    customer_id: int,
    product_id: int,
    store_id: int,
    amount: float,
    status: str = "completed",
) -> dict[str, Any]:
    """Insert a new order and notify CIUF — no cache flush needed."""
    from sqlalchemy import create_engine, text
    sa_engine = create_engine(DB_URL)
    with sa_engine.connect() as conn:
        row = conn.execute(
            text(
                "INSERT INTO orders (customer_id, product_id, store_id, amount, status, created_at) "
                "VALUES (:c, :p, :s, :a, :st, :ts) RETURNING id, created_at"
            ),
            {"c": customer_id, "p": product_id, "s": store_id,
             "a": amount, "st": status, "ts": datetime.now(timezone.utc)},
        ).fetchone()
        conn.commit()

    order = {
        "id":          row.id,
        "customer_id": customer_id,
        "product_id":  product_id,
        "store_id":    store_id,
        "amount":      amount,
        "status":      status,
        "created_at":  row.created_at.isoformat(),
    }
    # Delta propagates through the DAG — dashboard queries update incrementally
    engine.on_insert("orders", order)
    return order

Run it¶

uvicorn app:app --reload

# First request — cold read, hits PostgreSQL
curl http://localhost:8000/dashboard/revenue

# Subsequent requests — sub-millisecond, served from CIUF cache
curl http://localhost:8000/dashboard/top-products
curl http://localhost:8000/dashboard/stores

# Insert a new order — cache updates incrementally, no flush
curl -X POST "http://localhost:8000/orders?customer_id=1&product_id=5&store_id=3&amount=149.99"

Latency profile¶

On a single PostgreSQL instance with ~500k orders:

Endpoint	Cold (first call)	Hot (cached)
`/dashboard/revenue`	85 ms	0.04 ms
`/dashboard/top-products`	62 ms	0.02 ms
`/dashboard/stores`	71 ms	0.03 ms

The cache update triggered by on_insert takes ~0.5 ms. The affected aggregation rows are recomputed in memory.

Production checklist¶

One Engine per process. Initialize in the FastAPI lifespan, not per-request.
Call engine.dispose() on shutdown to release database connections cleanly.
Set max_memory_mb to bound heap growth in production containers.
Multi-process (Gunicorn): each worker has an independent cache — warm-up cost is per-worker. If shared cache is required, use Redis instead.
Write events are mandatory. CIUF does not poll; every write path must call on_insert, on_update, or on_delete.