create nested data structure easily
Project description
A small yet powerful tool to extend your pydantic schema with no pain.
It is the key to the realm of composition-oriented-development-pattern (wip)
[Attention] This package supports pydantic v2 only, it's incompatible with pydantic v1, if you want to play with pydantic v1, please use pydantic-resolve instead.
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 cases:
for more cases like:
- how to filter members
- how to make post calculation after resolved?
- and so on..
please read the following demos.
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 L47, L49
-
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
-
annotation_class:
class
if you have
from __future__ import annotation
, and pydantic raises error, use this config to update forward refsreference: test_25_parse_to_obj_for_pydantic_with_annotation.py, L39
-
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
model_config(default_required: bool)
-
default_required: if True, fields with default values will also in schema['required']
-
use with
Field(exclude=True)
to hide fields in schema and dumped resultreference: test_schema_config.py
Run FastAPI example:
FastAPI example
poetry shell
poetry install
cd examples
uvicorn fastapi_demo.main:app
visit 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 pydantic2_resolve-1.1.3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | a40a45b1d2b62abfa80a128fe5351947c367152fb15574d6f05f94aed6cf2a24 |
|
MD5 | 9222fdd6f0cc2236ec7567558cb551cd |
|
BLAKE2b-256 | 84395f08ed93c35086d29cbcfb431536fdb39b44bcde34068715a525ff5ff649 |