create nested data structure easily
Project description
A hierarchical solution for data fetching and processing
- use declaretive way to define view data, easy to maintain and develop
- use main query and loader query to break down complex queries, and better reuse
- provide various tools to precisely construct view data, no overhead
If you are using pydantic v2, please use pydantic2-resolve instead.
It is the key to composition-oriented-development-pattern (wip) https://github.com/allmonday/composition-oriented-development-pattern
Install
pip install pydantic-resolve
Concept
It taks 5 steps to convert from root data into view data.
- define schema of root data and descdants & load root data
- forward-resolve all descdants data
- backward-process the data
- tree shake the data marked as exclude=True
- get the output
How resolve works, level by level:
Quick start
Basic usage, resolve your fields. (N+1 query will happens in batch scenario)
Let's build a blog site with blogs and comments.
import asyncio
from pydantic import BaseModel
from pydantic_resolve import Resolver
comments_table = [
dict(id=1, blog_id=1, content='its interesting'),
dict(id=2, blog_id=1, content='i dont understand'),
dict(id=3, blog_id=2, content='why? how?'),
dict(id=4, blog_id=2, content='wow!')]
async def query_comments(blog_id: int):
print(f'run query - {blog_id}')
return [c for c in comments_table if c['blog_id'] == blog_id]
async def get_blogs():
return [
dict(id=1, title='what is pydantic-resolve'),
dict(id=2, title='what is composition oriented development pattarn'),
]
class Comment(BaseModel):
id: int
content: str
class Blog(BaseModel):
id: int
title: str
comments: list[Comment] = []
async def resolve_comments(self):
return await query_comments(self.id)
comment_count: int = 0
def post_comment_count(self):
return len(self.comments)
class MyBlogSite(BaseModel):
name: str
blogs: list[Blog] = []
async def resolve_blogs(self):
return await get_blogs()
comment_count: int = 0
def post_comment_count(self):
return sum([b.comment_count for b in self.blogs])
async def single():
blog = Blog(id=1, title='what is pydantic-resolve')
blog = await Resolver().resolve(blog)
print(blog)
async def batch():
my_blog_site = MyBlogSite(name: "tangkikodo's blog")
my_blog_site = await Resolver().resolve(my_blog_site)
print(my_blog_site.json(indent=4))
async def main():
await single()
# query_comments():
# >> run query - 1
await batch()
# query_comments(): Ooops!, N+1 happens
# >> run query - 1
# >> run query - 2
asyncio.run(main())
output of my_blog_site:
{
"name": "tangkikodo's blog",
"blogs": [
{
"id": 1,
"title": "what is pydantic-resolve",
"comments": [
{
"id": 1,
"content": "its interesting"
},
{
"id": 2,
"content": "i dont understand"
}
],
"comment_count": 2
},
{
"id": 2,
"title": "what is composition oriented development pattarn",
"comments": [
{
"id": 3,
"content": "why? how?"
},
{
"id": 4,
"content": "wow!"
}
],
"comment_count": 2
}
],
"comment_count": 4
}
Optimize N+1 with dataloader
- change
query_commentstoblog_to_comments_loader(loader function) - user loader in
resolve_comments
from pydantic_resolve import LoaderDepend
async def blog_to_comments_loader(blog_ids: list[int]):
print(blog_ids)
return build_list(comments_table, blog_ids, lambda c: c['blog_id'])
class Blog(BaseModel):
id: int
title: str
comments: list[Comment] = []
def resolve_comments(self, loader=LoaderDepend(blog_to_comments_loader)):
return loader.load(self.id)
comment_count: int = 0
def post_comment_count(self):
return len(self.comments)
# run main again
async def main():
await batch()
# blog_to_comments_loader():
# >> [1, 2]
# N+1 fixed
asyncio.run(main())
Simple demo:
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
Advanced demo:
https://github.com/allmonday/composition-oriented-development-pattern
API Reference
https://allmonday.github.io/pydantic-resolve/reference_api/
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
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.10.2.tar.gz
(14.7 kB
view hashes)
Built Distribution
Close
Hashes for pydantic_resolve-1.10.2-py3-none-any.whl
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | 1ec82d327bae85c0586410221d4ff32473b6991f03db4f77b3bb712aa7f3b6c5 |
|
| MD5 | 5392c79b0793ce10b75f4cae5d323197 |
|
| BLAKE2b-256 | 99ccb0c06dd3999b9210a27d00892fb06140b3ab4fde52a6a15dfb30265f60e4 |
