import asyncio
import time
from dataclasses import dataclass
from typing import Any, Dict
class _Miss:
def __repr__(self) -> str:
return "<MISS>"
MISS = _Miss()
@dataclass
class _Entry:
value: Any
expires_at: float
class TTLCache:
def __init__(self, purge_interval: float = 60.0) -> None:
self._data: Dict[str, _Entry] = {}
self._lock = asyncio.Lock()
self._purge_interval = purge_interval
self._last_purge = time.monotonic()
async def get(self, key: str) -> Any:
async with self._lock:
entry = self._data.get(key)
if entry is None:
return MISS
if entry.expires_at <= time.monotonic():
self._data.pop(key, None)
return MISS
return entry.value
async def set(self, key: str, value: Any, ttl: float) -> None:
async with self._lock:
self._data[key] = _Entry(value, time.monotonic() + ttl)
self._maybe_purge()
async def clear(self) -> None:
async with self._lock:
self._data.clear()
def _maybe_purge(self) -> None:
now = time.monotonic()
if now - self._last_purge < self._purge_interval:
return
self._last_purge = now
expired = [k for k, e in self._data.items() if e.expires_at <= now]
for k in expired:
self._data.pop(k, None)
import asyncio
from collections import defaultdict
from functools import wraps
from typing import Callable
from fastapi import Request
from .ttl_cache import MISS, TTLCache
cache = TTLCache()
_key_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
async def cache_key_builder(request: Request) -> str:
params = sorted(request.query_params.multi_items())
query = "&".join(f"{k}={v}" for k, v in params)
return f"{request.url.path}?{query}"
def cached(ttl: float) -> Callable:
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args, cache_key: str, **kwargs):
hit = await cache.get(cache_key)
if hit is not MISS:
return hit
# single-flight: only the first miss computes, others await it
async with _key_locks[cache_key]:
hit = await cache.get(cache_key)
if hit is not MISS:
return hit
result = await func(*args, **kwargs)
await cache.set(cache_key, result, ttl)
return result
return wrapper
return decorator
import asyncio
from fastapi import APIRouter, Depends
from .cache import cache, cache_key_builder, cached
router = APIRouter(prefix="/reports", tags=["reports"])
async def _aggregate(region: str, granularity: str) -> dict:
await asyncio.sleep(1.5) # stand-in for a slow upstream / heavy query
return {
"region": region,
"granularity": granularity,
"rows": [{"day": i, "total": i * 100} for i in range(7)],
}
@router.get("/summary")
@cached(ttl=30)
async def expensive_report(
region: str = "global",
granularity: str = "daily",
cache_key: str = Depends(cache_key_builder),
) -> dict:
return await _aggregate(region, granularity)
@router.delete("/summary/cache", status_code=204)
async def invalidate_report_cache() -> None:
await cache.clear()
This snippet shows how to add a small in-memory TTL cache to expensive FastAPI endpoints without pulling in Redis or another external store. It is useful when an endpoint recomputes the same result for many callers within a short window — for example an aggregated report or a slow upstream call — and the data can tolerate being a few seconds stale.
The TTLCache tab implements the store itself. Entries are kept in a plain dict keyed by a string, each holding the value and a monotonic expires_at timestamp computed from time.monotonic() so wall-clock jumps never corrupt expiry. Access is guarded by an asyncio.Lock because FastAPI serves coroutines on a single event loop and concurrent requests can interleave; the lock keeps get, set, and the opportunistic _purge sweep consistent. get returns a sentinel MISS object rather than None so that a legitimately cached None value is distinguishable from a miss.
The cache dependency tab is where the FastAPI idiom lives. cache_key_builder is a dependency that reads the request path and sorted query parameters and returns a stable string key — sorting matters so ?a=1&b=2 and ?b=2&a=1 collapse to one entry. cached is a decorator factory that wraps an endpoint coroutine: it resolves the injected key, checks the shared TTLCache, and either returns the hit or awaits the real handler and stores the result. The single-flight asyncio.Lock per key prevents a thundering herd where many simultaneous misses all recompute the same value; only the first computes while the rest await and read the freshly cached result.
The reports router tab wires it together on a realistic endpoint. expensive_report simulates a slow aggregation and is decorated with @cached(ttl=30), and the key dependency is declared with Depends, so the framework injects it exactly as it would any other dependency. A companion DELETE route calls cache.clear() to support manual invalidation.
The trade-offs are worth noting: an in-process cache is not shared across workers, so with several Uvicorn workers each holds its own copy and hit rates drop. It also grows unbounded unless purged, which the periodic sweep and TTL bound in practice. For a single-process service or a best-effort speedup it is a lightweight, dependency-free win; for coherence across a fleet a shared cache is the right tool.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
package com.example.myapp
import android.app.Application
import dagger.hilt.android.HiltAndroidApp
@HiltAndroidAppDependency injection with Hilt
Share this code
Here's the card — post it anywhere.
