pydantic-aiohttp 1.1.0

Creator: bradpython12

Last updated:

0 purchases

pydantic-aiohttp 1.1.0 Image
pydantic-aiohttp 1.1.0 Images
Add to Cart

Description:

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.

License

For personal and professional use. You cannot resell or redistribute these repositories in their original state.

Customer Reviews

There are no reviews.