create nested data structure easily
Project description
If you are using pydantic v2, please use pydantic2-resolve instead.
A small yet powerful tool to extend your pydantic schema, and then resolve all descendants automatically.
It's also the key to the realm of composable development pattern (wip)
What is composable pattarn? https://github.com/allmonday/composable-development-pattern
Install
pip install pydantic-resolve
Code snippets
- basic usage, resolve your fields.
import asyncio
from pydantic import BaseModel
from pydantic_resolve import Resolver
async def query_age(name):
print(f'query {name}')
await asyncio.sleep(1)
_map = {
'kikodo': 21,
'John': 14,
'老王': 40,
}
return _map.get(name)
class Person(BaseModel):
name: str
age: int = 0
async def resolve_age(self):
return await query_age(self.name)
is_adult: bool = False
def post_is_adult(self):
return self.age > 18
async def simple():
p = Person(name='kikodo')
p = await Resolver().resolve(p)
print(p)
# query kikodo
# Person(name='kikodo', age=21, is_adult=True)
people = [Person(name=n) for n in ['kikodo', 'John', '老王']]
people = await Resolver().resolve(people)
print(people)
# Oops!! the issue of N+1 query happens
# query kikodo
# query John
# query 老王
# [Person(name='kikodo', age=21, is_adult=True), Person(name='John', age=14, is_adult=False), Person(name='老王', age=40, is_adult=True)]
asyncio.run(simple())
- optimize
N+1
with dataloader
import asyncio
from typing import List
from pydantic import BaseModel
from pydantic_resolve import Resolver, LoaderDepend as LD
async def batch_person_age_loader(names: List[str]):
print(names)
_map = {
'kikodo': 21,
'John': 14,
'老王': 40,
}
return [_map.get(n) for n in names]
class Person(BaseModel):
name: str
age: int = 0
def resolve_age(self, loader=LD(batch_person_age_loader)):
return loader.load(self.name)
is_adult: bool = False
def post_is_adult(self):
return self.age > 18
async def simple():
people = [Person(name=n) for n in ['kikodo', 'John', '老王']]
people = await Resolver().resolve(people)
print(people)
# query query kikodo,John,老王 (N+1 query fixed)
# [Person(name='kikodo', age=21, is_adult=True), Person(name='John', age=14, is_adult=False), Person(name='老王', age=40, is_adult=True)]
asyncio.run(simple())
More examples:
cd examples
python -m readme_demo.0_basic
python -m readme_demo.1_filter
python -m readme_demo.2_post_methods
python -m readme_demo.3_context
python -m readme_demo.4_loader_instance
python -m readme_demo.5_subset
python -m readme_demo.6_mapper
python -m readme_demo.7_single
API
Resolver(loader_filters, global_loader_filter, loader_instances, ensure_type, context)
-
loader_filters:
dict
provide extra query filters along with loader key.
reference: 6_sqlalchemy_loaderdepend_global_filter.py L55, L59
-
global_loader_filter:
dict
provide global filter config for all dataloader instances
it will raise exception if some fields are duplicated with specific loader filter config in
loader_filters
reference: test_33_global_loader_filter.py L55, L59
-
loader_instances:
dict
provide pre-created loader instance, with can
prime
data into loader cache.reference: test_20_loader_instance.py, L62, L63
-
ensure_type:
bool
if
True
, resolve method is restricted to be annotated.reference: test_13_check_wrong_type.py
-
context:
dict
context can carry setting into each single resolver methods.
class Earth(BaseModel): humans: List[Human] = [] def resolve_humans(self, context): return [dict(name=f'man-{i}') for i in range(context['count'])] earth = await Resolver(context={'count': 10}).resolve(earth)
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:
list
, query result -
keys:
list
, batch_load_fn:keys -
fn:
lambda
, define the way to get primary keyhelper function to generate return value required by
batch_load_fn
. read the code for details.reference: test_utils.py, L32
mapper(param)
-
param:
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.reference: test_16_mapper.py
ensure_subset(base_class)
-
base_class:
class
it will raise exception if fields of decorated class has field not existed in
base_class
.reference: 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
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.9.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 168140a85d902b6a7a5f4a9708bfeb887afcc163646cbbb236aae3dcb28417d9 |
|
MD5 | da96b4e9e742ab9d39803b411feb4233 |
|
BLAKE2b-256 | 0565f6a469f6c9d1cd971741d797501da62d53a3405b65748bca0641079800ee |