Skip to main content

Simple HTTP Client based on aiohttp with integration of pydantic

Project description

pydantic_aiohttp - Symbiosis of Pydantic and Aiohttp

PyPI version shields.io PyPI pyversions PyPI license

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.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pydantic_aiohttp-1.0.1.tar.gz (12.5 kB view details)

Uploaded Source

Built Distribution

pydantic_aiohttp-1.0.1-py3-none-any.whl (15.4 kB view details)

Uploaded Python 3

File details

Details for the file pydantic_aiohttp-1.0.1.tar.gz.

File metadata

  • Download URL: pydantic_aiohttp-1.0.1.tar.gz
  • Upload date:
  • Size: 12.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.7.1 CPython/3.9.15 Linux/6.2.0-1018-azure

File hashes

Hashes for pydantic_aiohttp-1.0.1.tar.gz
Algorithm Hash digest
SHA256 129c9b74db36fbbbba5e248d0dd398cd0a5e28ef6016a6ed6f7e1d5820a6e66c
MD5 93b0e794d51846427ee903dcb5cac3ef
BLAKE2b-256 e1f7731c6a92bab4abf51bc69bf935bb676a846c988247ad4512c255d28bf505

See more details on using hashes here.

File details

Details for the file pydantic_aiohttp-1.0.1-py3-none-any.whl.

File metadata

  • Download URL: pydantic_aiohttp-1.0.1-py3-none-any.whl
  • Upload date:
  • Size: 15.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.7.1 CPython/3.9.15 Linux/6.2.0-1018-azure

File hashes

Hashes for pydantic_aiohttp-1.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 58d72518f042afcb70a42b331cd9dc3609ef0cc7445f64e117b67d4d308c5774
MD5 61ea74bf140e2d1a9c42f43c9540f4a8
BLAKE2b-256 869b9641457db3a2f42579fbe1759fddc741fc0348cbe21fa5433f4edc5e8d21

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page