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.1.0.tar.gz (13.5 kB view details)

Uploaded Source

Built Distribution

pydantic_aiohttp-1.1.0-py3-none-any.whl (13.4 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: pydantic_aiohttp-1.1.0.tar.gz
  • Upload date:
  • Size: 13.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/5.1.0 CPython/3.12.4

File hashes

Hashes for pydantic_aiohttp-1.1.0.tar.gz
Algorithm Hash digest
SHA256 73774ce75f41d8c122ee6f5b86feae8a2137418f0e8c18dad720baaa91b5f41c
MD5 d902e5c39c01d77496cca054595549e0
BLAKE2b-256 f897b04507295ad8d9ef16756d3551460fefc7f8041514263b3e47f11d02e9fb

See more details on using hashes here.

File details

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

File metadata

File hashes

Hashes for pydantic_aiohttp-1.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 5cc49cae264150ad7ab977bd4357ee4105ded397d2f5ab62571e088c72453961
MD5 74282b821dac663814340a9531a3da93
BLAKE2b-256 7b5d28e512355c2b956baf8c48e668709c4cfe92601902522445cbe1458ef40c

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