Quick Start
This page solves one endpoint-level problem with the smallest useful amount of code: each task has an owner_id, but the response model should expose a full owner object.
If you only need to fix a few N+1 issues in a handful of endpoints, this page and Core API may already be enough.
The Problem
Imagine a task list API that starts from data like this:
raw_tasks = [
{"id": 10, "title": "Design docs", "owner_id": 7},
{"id": 11, "title": "Refine examples", "owner_id": 8},
]
The response contract you actually want is not just owner_id. You want:
The naive implementation is usually a loop that fetches one owner per task. That is exactly the kind of N+1 problem pydantic-resolve is built to remove.
Install
If you later want MCP support as well:
The Smallest Useful Example
This example is self-contained and runnable. It uses a simple dict-based fake database so you can see the entire flow without setting up a real database.
import asyncio
from typing import Optional
from pydantic import BaseModel
from pydantic_resolve import Loader, Resolver, build_object
# --- Fake database ---
USERS = {
7: {"id": 7, "name": "Ada"},
8: {"id": 8, "name": "Bob"},
9: {"id": 9, "name": "Cara"},
}
# --- Loader function ---
async def user_loader(user_ids: list[int]):
"""Receives a batch of user_ids, returns results aligned with those keys."""
users = [USERS.get(uid) for uid in user_ids]
return build_object(users, user_ids, lambda user: user.id)
# --- Response models ---
class UserView(BaseModel):
id: int
name: str
class TaskView(BaseModel):
id: int
title: str
owner_id: int
owner: Optional[UserView] = None
def resolve_owner(self, loader=Loader(user_loader)):
return loader.load(self.owner_id)
# --- Resolve ---
raw_tasks = [
{"id": 10, "title": "Design docs", "owner_id": 7},
{"id": 11, "title": "Refine examples", "owner_id": 8},
]
tasks = [TaskView.model_validate(t) for t in raw_tasks]
tasks = await Resolver().resolve(tasks)
for t in tasks:
print(t.model_dump())
Output:
{'id': 10, 'title': 'Design docs', 'owner_id': 7, 'owner': {'id': 7, 'name': 'Ada'}}
{'id': 11, 'title': 'Refine examples', 'owner_id': 8, 'owner': {'id': 8, 'name': 'Bob'}}
What Each Piece Does
owner starts as None
The root task data does not include full owner objects, so the field starts empty. The resolver will fill it.
resolve_owner describes how to fetch the missing field
The method name follows the pattern resolve_<field_name>. The Loader(user_loader) argument declares a batched dependency — it does not call user_loader immediately.
user_loader receives all keys at once
async def user_loader(user_ids: list[int]):
users = [USERS.get(uid) for uid in user_ids]
return build_object(users, user_ids, lambda user: user.id)
The loader function receives a list of keys, not a single key. It must return results aligned with the incoming key order.
Resolver().resolve(tasks) walks the model tree
The resolver scans all model instances for resolve_* methods, collects the requested keys, calls each loader once per batch, and maps results back to the correct fields.
Why build_object Matters
user_loader must return a result for each key in user_ids, in the same order. build_object handles this alignment:
from pydantic_resolve import build_object
# build_object(items, keys, get_key_fn) -> list[item | None]
#
# Returns one element per key:
# - the matching item if found
# - None if no item matches that key
For one-to-many relationships (one sprint has many tasks), use build_list instead — it returns a list of lists.
Why This Avoids N+1
Suppose the task list contains 100 tasks. The resolver does not call user_loader 100 times. Instead:
- It collects all requested
owner_idvalues across all tasks. - It calls
user_loaderonce with the full batch:[7, 8, 7, 9, 8, ...]. - It maps each loaded user back to the right
TaskView.owner.
That is the core value of the library in its smallest form.
sequenceDiagram
participant R as Resolver
participant T as TaskView x100
participant L as user_loader
R->>T: scan resolve_owner methods
T->>R: collect owner_id values [7, 8, 7, 9, ...]
R->>L: user_loader([7, 8, 9]) ← one call
L->>R: [User7, User8, User9]
R->>T: assign owner to each task
Mental Model
The most useful first mental model is this:
resolve_*means: this field needs data from outside the current node.
Everything else in the library builds on that idea:
post_*runs after the subtree is readyExposeAs/SendTopass data across layersAutoLoadremoves the need to writeresolve_*at all
resolve_* Can Be Sync or Async
Both forms work:
# Sync — return a value directly
def resolve_owner(self, loader=Loader(user_loader)):
return loader.load(self.owner_id)
# Async — await the loader, then transform the result
async def resolve_owner(self, loader=Loader(user_loader)):
user = await loader.load(self.owner_id)
return user
Use async when you need to post-process the loaded data before assignment.
When to Stop Here
Staying at this level is completely reasonable when:
- you only need to fix a few related-data fields
- your response models are still changing quickly
- you do not have repeated relationship wiring across many models yet
Next
Continue to Core API to extend the same pattern from one field to a nested tree: Sprint -> Task -> User.