chinese - latest

discord

english version is under revamp.

When building APIs for relational data, we are always faced with layers of nested data and ever-changing business requirements.

Is there a solution that combines flexibility, performance, and maintainability?

This solution should support the following features:

• Asynchronous
• The definition of multi-layer data structures and can easily expand related data
• Global context, scoped context
• The ability for each layer to have a hook to operate the data after the descendant data are resolved
• Pick the required fields
• Avoid N+1 queries
• Convenient for debugging

This repo will implement such a solution through a series of examples, with pydantic2-resolve and some conventions.

• https://github.com/allmonday/pydantic-resolve
• https://github.com/allmonday/pydantic2-resolve

A mini-jira system. (ERD)

---
title: Mini JIRA
---

erDiagram
    Team ||--o{ Sprint : one_to_many
    Team ||--o{ User : one_to_many
    Sprint ||--o{ Story : one_to_many
    Story ||--o{ Task : one_to_many
    Story ||--|| User : one_to_one
    Task ||--|| User : one_to_one

    Team {
      int id
      string name
    }
    Sprint {
      int id
      string name
    }
    Story {
      int id
      int sprint_id
      string name
      int owner_id
    }
    Task {
      int id
      int story_id
      string name
      int owner_id
    }
    User {
      int id
      string name
    }

In daily development, in order to obtain data with complex structures, we often choose to assemble the data after multiple requests from the client, or construct complex queries in the service.

If requirements changes, the query at the client or service layer will also need to be adjusted accordingly.

This change pollutes the ideal layered design and make changes happen in the service/client.

The current popular approach is using GraphQL, but the introduction cost of the entire solution is not low for the backend. Additionally, the frontend needs to manually write queries to describe fields, which lacks the smooth experience of directly using RPC-like requests.

The combination-oriented development pattern is to solve this problem.

By using pydantic2-resolve, and some conventions.

The router layer is responsible for building the schema to encapsulate changes, thereby avoiding changes in service and client.

In the cases of this repo, there are two directories:

Services and Routers.

Services are mainly responsible for a certain business service, it includes:

• Schema definition
• Business query
• Dataloaders

Routers return the required data by combining [query, schema, loader] from multiple services.

The concept router here is mostly like controller, handling the composition of schemas is it's core function

This demo is based on FastAPI, so I'll call it router instead.

This composition approach enables the flexible combination of generic services with specific business logic, allowing the service to quickly and concisely construct routers/APIs that meet the requirements of the business, and keeps services stable (unchanged) at the same time.

For example, in the example, Sample1StoryDetail is composed of multiple schemas + loaders. The Story data inherited by Sample1StoryDetail is provided by the business query.

compare to GraphQL, you'll have full control over dataloader.

from typing import Optional
from pydantic_resolve import LoaderDepend

# loaders
import src.services.task.loader as tl
import src.services.user.loader as ul
import src.services.story.loader as sl
import src.services.sprint.loader as spl

# schemas
import src.services.story.schema as ss
import src.services.task.schema as ts
import src.services.user.schema as us
import src.services.sprint.schema as sps
import src.services.team.schema as tms

# compose together
class Sample1TaskDetail(ts.Task):
    # inherit and extend fields
    user: Optional[us.User] = None
    def resolve_user(self, loader=LoaderDepend(ul.user_batch_loader)):
        return loader.load(self.owner_id)

class Sample1StoryDetail(ss.Story):
    # inherit and extend fields
    tasks: list[Sample1TaskDetail] = []
    def resolve_tasks(self, loader=LoaderDepend(tl.story_to_task_loader)):
        return loader.load(self.id)

    owner: Optional[us.User] = None
    def resolve_owner(self, loader=LoaderDepend(ul.user_batch_loader)):
        return loader.load(self.owner_id)

# query
@route.get('/stories-with-detail', response_model=List[Sample1StoryDetail])
async def get_stories_with_detail(session: AsyncSession = Depends(db.get_session)):
    # fetching root data from query
    stories = await sq.get_stories(session)

    # load it into Resolvable schema
    stories = [Sample1StoryDetail.model_validate(t) for t in stories]

    # resolve it
    stories = await Resolver().resolve(stories)

    # done
    return stories

OpenAPI, aka Swagger

python -m venv venv
source venv/bin/activate
pip install -r requirement.txt
uvicorn src.main:app --port=8000 --reload
# http://localhost:8000/docs

You can execute it in swagger to view the return value of each API

with UI

python -m venv venv
source venv/bin/activate
pip install -r requirement.txt
uvicorn src.main:app  --port=8001 --reload

cd fe-demo
npm install
# generate sdk
npm run generate-client

npm run dev

or visit the interactive UI.

• Example 1: Construction of multi-layer nested structures
• Example 2: Data filtering in Loader
• Example 3: Exposing fields to descendant nodes
• Example 4: After resolved, post-processing of obtaining data
• Example 5: Reuse of Loaders
• Example 6: Picking the fields to be returned
• Example 7: Loader instance
• Service testing is enough
• Compare to GraphQL wip
• generating ts-sdk by openapi-typescirpt-codegen wip