pydantic-resolve 1.1.1

Make pydantic have a GraphQL-like assembly experience.

Pydantic-resolve

A small yet powerful package which can run resolvers to generate deep nested datasets.

example:

# define loader functions
async def friends_batch_load_fn(names):
    mock_db = {
        'tangkikodo': ['tom', 'jerry'],
        'john': ['mike', 'wallace'],
        'trump': ['sam', 'jim'],
        'sally': ['sindy', 'lydia'],
    }
    return [mock_db.get(name, []) for name in names]

async def contact_batch_load_fn(names):
    mock_db = {
        'tom': 100, 'jerry':200, 'mike': 3000, 'wallace': 400, 'sam': 500,
        'jim': 600, 'sindy': 700, 'lydia': 800, 'tangkikodo': 900, 'john': 1000,
        'trump': 1200, 'sally': 1300,
    }
    return [mock_db.get(name, None) for name in names]

# define schemas
class Contact(BaseModel):
    number: Optional[int]

class Friend(BaseModel):
    name: str

    contact: int = 0
    @mapper(lambda n: Contact(number=n))
    def resolve_contact(self, loader=LoaderDepend(contact_batch_load_fn)):
        return loader.load(self.name)

class User(BaseModel):
    name: str
    age: int

    greeting: str = ''
    def resolve_greeting(self):
        return f"hello, i'm {self.name}, {self.age} years old."

    contact: int = 0
    @mapper(lambda n: Contact(number=n))
    def resolve_contact(self, loader=LoaderDepend(contact_batch_load_fn)):
        return loader.load(self.name)

    friends: List[Friend] = []
    @mapper(lambda items: [Friend(name=item) for item in items])  # transform after data received
    def resolve_friends(self, loader=LoaderDepend(friends_batch_load_fn)):
        return loader.load(self.name)

class Root(BaseModel):
    users: List[User] = []
    def resolve_users(self):
        return [
          {"name": "tangkikodo", "age": 19},
            User(name="tangkikodo", age=19),  # transform first
            User(name='john', age=21),
            # User(name='trump', age=59),
            # User(name='sally', age=21),
            # User(name='some one', age=0)
        ]

async def main():
    import json
    root = await Resolver().resolve(Root())
    dct = root.dict()
    print(json.dumps(dct, indent=4))

asyncio.run(main())

output:

{
  "users": [
    {
      "name": "tangkikodo",
      "age": 19,
      "greeting": "hello, i'm tangkikodo, 19 years old.",
      "contact": {
        "number": 900
      },
      "friends": [
        {
          "name": "tom",
          "contact": {
            "number": 100
          }
        },
        {
          "name": "jerry",
          "contact": {
            "number": 200
          }
        }
      ]
    },
    {
      "name": "john",
      "age": 21,
      "greeting": "hello, i'm john, 21 years old.",
      "contact": {
        "number": 1000
      },
      "friends": [
        {
          "name": "mike",
          "contact": {
            "number": 3000
          }
        },
        {
          "name": "wallace",
          "contact": {
            "number": 400
          }
        }
      ]
    }
  ]
}

• Full-feature example which includes dataloader, LoaderDepend and global loader_filters
• Helps you asynchoronously, resursively resolve a pydantic object (or dataclass object)
• When used in conjunction with aiodataloader, allows you to easily generate nested data structures without worrying about generating N+1 queries.
• say byebye to contextvars when using dataloader.
• Inspired by GraphQL and graphene

Install

pip install pydantic-resolve

pip install "pydantic-resolve[dataloader]"  # install with aiodataloader, from v1.0, aiodataloader is a default dependency, [dataloader] is removed.

• use resolve for simple scenario,
• use Resolver and LoaderDepend for complicated nested batch query.

from pydantic_resolve import (
    resolve,                     # handle simple resolving task
    Resolver, LoaderDepend,      # handle schema resolving with LoaderDepend and DataLoader
    ResolverTargetAttrNotFound, DataloaderDependCantBeResolved, LoaderFieldNotProvidedError
)

Run FastAPI example

poetry shell
cd examples
uvicorn fastapi_demo.main:app
# http://localhost:8000/docs#/default/get_tasks_tasks_get

Some documentations.

• Reason
• How LoaderDepend works
• Comparsion with common solutions

For more examples, please explore examples folder.

Unittest

poetry run python -m unittest  # or
poetry run pytest  # or
poetry run tox

Coverage

poetry run coverage run -m pytest
poetry run coverage report -m