pydantic-resolve

pydantic-resolve is a framework designed to bring Clean Architecture to Python web development. It enables developers to define business entities and declare relationships using an Entity-Relationship Diagram (ERD). The framework then automatically assembles nested data trees and eliminates N+1 query problems through its Resolver and Loader mechanisms.

Key capabilities include:

• Automatic Data Assembly: Uses resolve_* methods and Loader to batch lookups and map results back to nested structures.
• Derived Field Computation: Uses post_* methods to compute fields after descendant nodes are fully resolved.
• Cross-Layer Data Flow: Uses ExposeAs, SendTo, and Collector to pass context down or aggregate data up the tree.
• Centralized Relationships: Uses ER Diagrams and AutoLoad to reuse relationship wiring across multiple models, which can also power GraphQL and MCP services.

Entity-First architecture is a design pattern that decouples business logic from database implementation. Instead of building API schemas directly from database models (ORM-first), you define stable Business Entities (Entities) and their Relationships (ERD) first.

Once Entities are defined, you can derive multiple API response structures (Use Cases) by selecting subsets of fields or adding extended/computed fields. This approach provides:

• Stability: Database changes (like splitting tables) only require updating the data-loading layer, leaving the domain and API layers untouched.
• Clear Semantics: Entities use business language (e.g., owner) rather than database terminology (e.g., owner_id).
• Multi-source Abstraction: Entities can declare relationships that span different data sources (e.g., one field from PostgreSQL, another from a Redis cache or an RPC call) without the consumer knowing the source.

Entity-First architecture is a design pattern that separates business logic from data access and API contracts. It is composed of three layers:

1. Domain Layer (Entity & ERD): Defines business concepts (e.g., UserEntity) and their relationships (e.g., Task.owner -> UserEntity) using Pydantic. This layer is independent of the database or framework.
2. API Layer (Response): Defines specific use-case data structures by selecting subsets of Entities and extending them with computed or associated fields (e.g., TaskResponse).
3. Data Layer (Loader): Encapsulates persistence details (ORM, RPC, Cache, or HTTP API) and provides a unified interface for loading data via batching.

This separation allows the database structure, business logic, and API contracts to evolve independently without breaking the entire system.

graph TD
    subgraph DATA["Data Query Details"]
        D1["Encapsulate persistence details<br/>ORM / RPC / Cache / HTTP API"]
        D2["Unified data loading interface (Loader)"]
        D3["Batch loading, caching, error handling"]
        D4["Example: user_loader, task_to_owner_loader"]
    end

    subgraph DOMAIN["Business Model and Relationships"]
        subgraph ENTITY["Entity (Business Entity)"]
            E1["Define business concepts and fields"]
            E2["Independent of database and framework"]
            E3["Example: UserEntity, TaskEntity, TeamEntity"]
        end

        subgraph ERD["ERD (Entity Relationship Diagram)"]
            R1["Declare relationships between entities"]
            R2["Define how to load associated data"]
            R3["Example: Task.owner -> UserEntity"]
        end
    end

    subgraph API["API Contract (Use Case Composition)"]
        A1["Select fields from Entity"]
        A2["Define API-specific computed fields"]
        A3["Declare how to use associated data"]
        A4["Example: TaskResponse, UserSummary"]
    end

    API -->|depends on| DOMAIN
    DOMAIN -->|depends on| DATA

    style DATA fill:#e1f5fe,stroke:#01579b,stroke-width:2px
    style DOMAIN fill:#f3e5f5,stroke:#4a148c,stroke-width:2px
    style API fill:#fff3e0,stroke:#e65100,stroke-width:2px
    style ENTITY fill:#ffffff,stroke:#4a148c,stroke-width:1px
    style ERD fill:#ffffff,stroke:#4a148c,stroke-width:1px

By default, if multiple fields share a DataLoader but require different response types (e.g., a lightweight TaskCard vs. a heavy TaskDetail), the Resolver creates one shared loader. This loader's _query_meta.fields becomes the union of all required fields, causing lightweight views to fetch unnecessary data.

Setting split_loader_by_type=True creates independent DataLoader instances for each request_type. Each instance's _query_meta will only contain the columns actually needed for that specific type.

Trade-offs:

• Performance: Split loaders maintain independent caches. If two different types request the same ID, the database will be queried twice.
• Incompatibility: You cannot use split_loader_by_type=True if you are also providing loader_instances (pre-created instances).

class Dashboard(BaseModel):
    id: int
    cards: List[TaskCard] = []       # Needs: id, title
    details: List[TaskDetail] = []   # Needs: id, title, desc, status, ...

    def resolve_cards(self, loader=Loader(TaskLoader)):
        return loader.load(self.id)

    def resolve_details(self, loader=Loader(TaskLoader)):
        return loader.load(self.id)

# Use split_loader_by_type=True to ensure 'cards' doesn't fetch 'details' columns
result = await Resolver(split_loader_by_type=True).resolve(Dashboard(id=1))

By default, if multiple fields share a DataLoader but request different response types (e.g., a lightweight TaskCard vs a heavy TaskDetail), the Resolver creates a single shared loader instance. This instance's _query_meta.fields becomes the union of all requested columns, causing lightweight views to fetch unnecessary data.

Setting split_loader_by_type=True creates an independent DataLoader instance per request_type. This ensures each instance's _query_meta only contains the columns required for its specific target type.

Limitations:

• Cache Fragmentation: Split loaders maintain independent caches. If two different types request the same ID, the database will be queried twice.
• Incompatibility: You cannot use split_loader_by_type=True if you are also providing loader_instances.

class Dashboard(BaseModel):
    id: int
    cards: List[TaskCard] = []       # only needs id, title
    details: List[TaskDetail] = []   # needs id, title, desc, status, ...

    def resolve_cards(self, loader=Loader(TaskLoader)):
        return loader.load(self.id)

    def resolve_details(self, loader=Loader(TaskLoader)):
        return loader.load(self.id)

# Use split_loader_by_type=True to prevent 'cards' from fetching 'details' columns
result = await Resolver(split_loader_by_type=True).resolve(Dashboard(id=1))

To make a node's data available to all its descendants, define the __pydantic_resolve_expose__ dictionary on the parent model.

• The key is the field name to expose.
• The value is a globally unique alias.

Descendants can access this data in both resolve and post methods via the ancestor_context parameter using the unique alias.

from pydantic import BaseModel
from pydantic_resolve import Resolver

class Owner(BaseModel):
    # Expose 'name' to descendants using the alias 'owner_name'
    __pydantic_resolve_expose__ = { 'name': 'owner_name' }
    name: str
    items: List[Item]

class Item(BaseModel):
    name: str
    description: str = ''

    def post_description(self, ancestor_context):
        # Access the exposed data via the alias
        return f"this is item: {self.name}, it belongs to {ancestor_context['owner_name']}"

owners = [dict(name="alice", items=[dict(name='car')])]
owners = await Resolver().resolve([Owner(**o) for o in owners])

When deciding between standard SQLAlchemy eager loading and pydantic-resolve, consider the following trade-offs:

Recommendation:

• Use SQLAlchemy Eager Loading for simple cases (e.g., 1-2 levels of nesting, single endpoint).
• Use pydantic-resolve when you have deep nesting (3-4+ levels) or need multiple variants (list view, detail view, analytics view) of the same data, as it allows for significant loader and schema reuse.

The pydantic-resolve development workflow is divided into three distinct phases, moving from pure data models to full API exposure.

Layer Responsibilities

• methods.py: Acts as the pure business logic layer. It returns ORM Models.
• service.py: Responsible for DTO conversion. It calls methods.py, performs the Resolver().resolve() call, and returns DTOs.

Introduced in v2, ErDiagram lets you define an application-layer Entity Relationship Diagram. This allows you to specify Relationship objects that include a default dataloader, removing the need to manually write resolve_* methods for every field.

There are two ways to define a diagram:

1. Standalone Definition

Create an ErDiagram instance containing Entity objects. Each Entity defines its class (kls) and a list of Relationship objects.

2. Declarative Definition (via base_entity)

Use base_entity() to create a base class. Define relationships within the __relationships__ attribute of your models. You can then retrieve the diagram using BaseEntity.get_diagram().

Once a diagram is defined, you can use AutoLoad() to automatically wire up the relationships in your models.

from pydantic_resolve import base_entity, Relationship

# Declarative approach
BaseEntity = base_entity()

class Story(BaseModel, BaseEntity):
    __relationships__ = [
        Relationship(fk='id', target=list[Task], name='tasks', loader=task_loader.story_to_task_loader)
    ]
    id: int
    name: str

diagram = BaseEntity.get_diagram()