0 purchases
pydanticaiohttp 1.1.0
pydantic_aiohttp - Symbiosis of Pydantic and Aiohttp
This repository provides simple HTTP Client based on aiohttp with integration of pydantic
Examples
Basic example
import asyncio
import pydantic
from pydantic_aiohttp import Client
from pydantic_aiohttp.responses import (
JSONResponseClass,
PlainTextResponseClass,
PydanticModelResponseClass
)
class Todo(pydantic.BaseModel):
userId: int
id: int
title: str
completed: bool
async def main():
client = Client('https://jsonplaceholder.typicode.com')
async with client:
# Text response
todo = await client.get('/todos/1', response_class=PlainTextResponseClass)
print(isinstance(todo, str)) # True
# JSON Response
todo = await client.get('/todos/1', response_class=JSONResponseClass)
print(isinstance(todo, dict)) # True
# You can achieve the same result if you know exact shape of response, dict for example
todo = await client.get('/todos/1', response_class=PydanticModelResponseClass, response_model=dict)
print(isinstance(todo, dict)) # True
# Deserialization in pydantic model
todo = await client.get('/todos/1', response_class=PydanticModelResponseClass, response_model=Todo)
print(isinstance(todo, Todo)) # True
# PydanticModelResponseClass is used by default, so you can omit it
todo = await client.get('/todos/1', response_model=Todo)
print(isinstance(todo, Todo)) # True
if __name__ == '__main__':
asyncio.run(main())
Explicitly close connection
import asyncio
import pydantic
from pydantic_aiohttp import Client
class Todo(pydantic.BaseModel):
userId: int
id: int
title: str
completed: bool
async def main():
client = Client('https://jsonplaceholder.typicode.com')
try:
await client.get('/todos/1', response_model=Todo)
finally:
# Don't forget to close client session after use
await client.close()
if __name__ == '__main__':
asyncio.run(main())
Downloading files
import asyncio
import uuid
from pydantic_aiohttp import Client
async def main():
client = Client('https://source.unsplash.com')
async with client:
filepath = await client.download_file("/random", filepath=f"random_{uuid.uuid4()}.jpg")
print(filepath)
if __name__ == '__main__':
asyncio.run(main())
Handling errors parsed as pydantic models
import http
import asyncio
import pydantic
import pydantic_aiohttp
from pydantic_aiohttp import Client
class FastAPIValidationError(pydantic.BaseModel):
loc: list[str]
msg: str
type: str
class FastAPIUnprocessableEntityError(pydantic.BaseModel):
detail: list[FastAPIValidationError]
class User(pydantic.BaseModel):
id: str
email: str
first_name: str
last_name: str
is_admin: bool
async def main():
client = Client(
"https://fastapi.example.com",
error_response_models={
http.HTTPStatus.UNPROCESSABLE_ENTITY: FastAPIUnprocessableEntityError
}
)
try:
# Imagine, that "email" field is required for this route
await client.post(
"/users",
body={
"first_name": "John",
"last_name": "Doe"
},
response_model=User
)
except pydantic_aiohttp.HTTPUnprocessableEntity as e:
# response field of exception now contain parsed pydantic model entity
print(e.response.detail[0].model_dump_json(indent=4))
# >>>
# {
# "loc": [
# "body",
# "email"
# ],
# "msg": "field required",
# "type": "value_error.missing"
# }
finally:
await client.close()
if __name__ == '__main__':
asyncio.run(main())
LICENSE
This project is licensed under the terms of the MIT license.
For personal and professional use. You cannot resell or redistribute these repositories in their original state.
There are no reviews.