from typing import List, Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from .dataloader import DataLoader
from .models import User
def build_user_loader(session: AsyncSession) -> DataLoader:
async def load_users_by_ids(ids: List[int]) -> List[Optional[User]]:
result = await session.execute(
select(User).where(User.id.in_(ids))
)
by_id = {user.id: user for user in result.scalars().all()}
# Preserve the requested order; None for ids with no matching row.
return [by_id.get(uid) for uid in ids]
return DataLoader(load_users_by_ids)
import asyncio
from typing import Any, Awaitable, Callable, Hashable, List, Optional
BatchFn = Callable[[List[Hashable]], Awaitable[List[Optional[Any]]]]
class DataLoader:
def __init__(self, batch_fn: BatchFn) -> None:
self._batch_fn = batch_fn
self._queue: List[Hashable] = []
self._futures: dict[Hashable, asyncio.Future] = {}
self._cache: dict[Hashable, Any] = {}
self._scheduled = False
def load(self, key: Hashable) -> asyncio.Future:
if key in self._cache:
fut = asyncio.get_event_loop().create_future()
fut.set_result(self._cache[key])
return fut
if key in self._futures:
return self._futures[key]
loop = asyncio.get_event_loop()
fut = loop.create_future()
self._futures[key] = fut
self._queue.append(key)
if not self._scheduled:
self._scheduled = True
loop.call_soon(lambda: asyncio.ensure_future(self._dispatch()))
return fut
async def _dispatch(self) -> None:
keys = list(dict.fromkeys(self._queue))
self._queue.clear()
self._scheduled = False
try:
values = await self._batch_fn(keys)
except Exception as exc: # propagate to every waiting caller
for key in keys:
self._futures.pop(key).set_exception(exc)
return
for key, value in zip(keys, values):
self._cache[key] = value
self._futures.pop(key).set_result(value)
import asyncio
from typing import List
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from .dataloader import DataLoader
from .db import get_session
from .loaders import build_user_loader
from .models import Comment
from .schemas import CommentOut, UserOut
router = APIRouter()
def get_user_loader(session: AsyncSession = Depends(get_session)) -> DataLoader:
return build_user_loader(session)
@router.get("/posts/{post_id}/comments", response_model=List[CommentOut])
async def list_comments(
post_id: int,
session: AsyncSession = Depends(get_session),
user_loader: DataLoader = Depends(get_user_loader),
):
result = await session.execute(
select(Comment).where(Comment.post_id == post_id).order_by(Comment.created_at)
)
comments = result.scalars().all()
authors = await asyncio.gather(
*(user_loader.load(c.author_id) for c in comments)
)
return [
CommentOut(
id=c.id,
body=c.body,
author=UserOut.model_validate(author) if author else None,
)
for c, author in zip(comments, authors)
]
Serializing a list of resources that each reference a related record is a classic source of N+1 queries: rendering 50 comments triggers 50 separate author lookups. The DataLoader pattern fixes this by collecting the individual key requests made during one response, coalescing them into a single batched query, and handing each caller back only the row it asked for. This snippet wires that pattern into an async FastAPI endpoint so the author of each comment is joined in with exactly one extra query.
In dataloader.py, DataLoader buffers keys instead of querying immediately. Each call to load appends the key to _queue, stores an asyncio.Future, and — crucially — schedules _dispatch with loop.call_soon only once per pending batch. Because call_soon runs after the current synchronous chunk of the event loop yields, every load issued while serializing the response lands in the same queue before _dispatch fires. _dispatch deduplicates keys, calls the user-supplied batch_fn a single time, then resolves each future with its matching value or None. The _scheduled flag guards against dispatching twice, and results are cached in _cache so repeated keys within a request are free.
In loaders.py, load_users_by_ids is the concrete batch_fn: it runs one SELECT ... WHERE users.id IN (...), builds an id-to-row map, and returns values ordered to match the requested keys. Returning None for missing ids keeps the loader contract total, so a caller never hangs waiting on a key that has no row. build_user_loader creates a fresh loader per request, which matters because the cache and in-flight batch must not leak between users.
In routes.py, list_comments fetches the comments, then awaits user_loader.load(...) for each one. asyncio.gather lets those loads run concurrently while the loader still collapses them into one query. The per-request loader is created inside the dependency get_user_loader so its lifecycle matches the request. The trade-off is added indirection and a reliance on event-loop timing, but it removes the N+1 without threading join logic through every query. It shines when the related resource is reused across many parents or reached through several code paths.
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
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
Rails.application.configure do
config.after_initialize do
Bullet.enable = true
Bullet.alert = false
Bullet.bullet_logger = true
Bullet.console = trueN+1 query detection with Bullet gem
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
Share this code
Here's the card — post it anywhere.
