Make pydantic have a GraphQL-like assembly experience.
Project description
Pydantic-resolve
A small yet powerful package which can run resolvers to generate deep nested datasets.
example:
import asyncio
from typing import List, Optional
from pydantic import BaseModel
from pydantic_resolve import Resolver, mapper, LoaderDepend
# define dataset and 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': 1001, 'jerry': 1002, 'mike': 1003, 'wallace': 1004, 'sam': 1005,
'jim': 1006, 'sindy': 1007, 'lydia': 1008, 'tangkikodo': 1009, 'john': 1010,
'trump': 2011, 'sally': 2012,
}
result = []
for name in names:
n = mock_db.get(name, None)
result.append({'number': n} if n else None)
return result
# define data schemas
class Contact(BaseModel):
number: Optional[int]
class Friend(BaseModel):
name: str
contact: Optional[Contact] = None
@mapper(Contact) # 1. resolve dataloader and map return dict to Contact object
def resolve_contact(self, contact_loader=LoaderDepend(contact_batch_load_fn)):
return contact_loader.load(self.name)
is_contact_10: bool = False
def post_is_contact_10(self): # 3. after resolve_contact executed, do extra computation
if self.contact:
if str(self.contact.number).startswith('10'):
self.is_contact_10 = True
else:
self.is_contact_10 = False
class User(BaseModel):
name: str
age: int
greeting: str = ''
async def resolve_greeting(self):
await asyncio.sleep(1)
return f"hello, i'm {self.name}, {self.age} years old."
contact: Optional[Contact] = None
@mapper(Contact)
def resolve_contact(self, contact_loader=LoaderDepend(contact_batch_load_fn)):
return contact_loader.load(self.name)
friends: List[Friend] = []
@mapper(lambda names: [Friend(name=name) for name in names])
def resolve_friends(self, friend_loader=LoaderDepend(friends_batch_load_fn)):
return friend_loader.load(self.name)
friend_count: int = 0
def post_friend_count(self):
self.friend_count = len(self.friends)
class Root(BaseModel):
users: List[User] = []
@mapper(lambda items: [User(**item) for item in items])
def resolve_users(self):
return [
{"name": "tangkikodo", "age": 19},
{"name": "john", "age": 20},
{"name": "trump", "age": 21},
{"name": "sally", "age": 22},
{"name": "no man", "age": 23},
]
async def main():
import json
root = Root()
root = await Resolver().resolve(root) # 4. run it
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": 1009
},
"friends": [
{
"name": "tom",
"contact": {
"number": 1001
},
"is_contact_10": true
},
{
"name": "jerry",
"contact": {
"number": 1002
},
"is_contact_10": true
}
],
"friend_count": 2
},
{
"name": "john",
"age": 20,
"greeting": "hello, i'm john, 20 years old.",
"contact": {
"number": 1010
},
"friends": [
{
"name": "mike",
"contact": {
"number": 1003
},
"is_contact_10": true
},
{
"name": "wallace",
"contact": {
"number": 1004
},
"is_contact_10": true
}
],
"friend_count": 2
},
{
"name": "trump",
"age": 21,
"greeting": "hello, i'm trump, 21 years old.",
"contact": {
"number": 2011
},
"friends": [
{
"name": "sam",
"contact": {
"number": 1005
},
"is_contact_10": true
},
{
"name": "jim",
"contact": {
"number": 1006
},
"is_contact_10": true
}
],
"friend_count": 2
},
{
"name": "sally",
"age": 22,
"greeting": "hello, i'm sally, 22 years old.",
"contact": {
"number": 2012
},
"friends": [
{
"name": "sindy",
"contact": {
"number": 1007
},
"is_contact_10": true
},
{
"name": "lydia",
"contact": {
"number": 1008
},
"is_contact_10": true
}
],
"friend_count": 2
},
{
"name": "no man",
"age": 23,
"greeting": "hello, i'm no man, 23 years old.",
"contact": null,
"friends": [],
"friend_count": 0
}
]
}
Install
pip install pydantic-resolve
- use
resolve
for simple scenario, - use
Resolver
andLoaderDepend
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 # errors
)
Run FastAPI example
poetry shell
cd examples
uvicorn fastapi_demo.main:app
# http://localhost:8000/docs#/default/get_tasks_tasks_get
Some documentations.
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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
pydantic_resolve-1.3.1.tar.gz
(8.0 kB
view hashes)
Built Distribution
Close
Hashes for pydantic_resolve-1.3.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 9309b75a48cba03648108a5dbd545150cf5fc31389e66e366b79a85157ea4556 |
|
MD5 | cc6c8eeae0516c3e73cd9f22cb99d8c3 |
|
BLAKE2b-256 | 6035c477ba2b1f6b606062aaa469f524bcb5aef944de313b3694e3c037ab3daa |