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:
# prepare dataloader
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
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
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]) # manually convert
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] = []
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},
]
# resolve results
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": "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.
API
Resolver(loader_filters, loader_instances, ensure_type)
-
loader_filters
provide extra query filters along with loader key.
detail:
examples/6_sqlalchemy_loaderdepend_global_filter.py
L55, L59 -
loader_instances
provide pre-created loader instance, with can
prime
data into loader cache.detail:
tests/resolver/test_20_loader_instance.py
, L62, L63 -
ensure_type
if
True
, resolve method is restricted to be annotated.detail:
tests/resolver/test_13_check_wrong_type
LoaderDepend(loader_fn)
-
loader_fn
: subclass of DataLoader or batch_load_fn. detaildeclare dataloader dependency,
pydantic-resolve
will take the care of lifecycle of dataloader.
build_list(rows, keys, fn)
& build_object(rows, keys, fn)
-
rows
: query result -
keys
: batch_load_fn:keys -
fn
: define the way to get primary keyhelper function to generate return value required by
batch_load_fn
. read the code for details.
mapper(param)
-
param
: can be either a class of pydantic or dataclass, or a lambda.pydantic-resolve
will trigger the fn inmapper
after inner future is resolved. it exposes an interface to change return schema even from the same dataloader. if param is a class, it will try to automatically transform it.
ensure_subset(base_class)
-
base_class
: pydantic classit can raise exception if fields of decorated class has field not existed in
base_class
.detail:
tests/utils/test_2_ensure_subset.py
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
Built Distribution
Hashes for pydantic_resolve-1.5.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 19c481957e174fc3b108324ebf63f54bf3b30c51d79ec721d508d6d538cf18e2 |
|
MD5 | 35fae2dad60bf4a169f22d9be21d25ea |
|
BLAKE2b-256 | 2f23a2b384d1f5f01c8806e50637d199f47df23624812f2315897d8f668ae5da |