unihttp – пишем декларативные API-клиенты на Python

Привет! Хочу представить вам свою библиотеку – unihttp.

Уверен, что все сталкивались с необходимостью работы с какими-либо API, но что делать, если у этого сервиса (внешнего или внутреннего) отсутствует библиотека, позволяющая лаконично вызывать нужные методы?

Если это единичный вызов – можно обернуть в функцию и используя голые requests / aiohttp отправлять запросы. В случае, если у вас идет тесное взаимодействие с этим API, или у вас приложение в принципе построено для работы с ним, то ваш код превращается в огромную простыню функций или вызовов requests / aiohttp и последующего трансформирования ответа в нужный тип.

Рано или поздно, вы приходите к мысли о том, что хорошей идеей было бы вынести работу с этим API в библиотеку. Появляется новый вопрос: как ее делать?

В основном, мы решаем следующие вопросы в процессе размышления над библиотекой:

1. Каким образом мы должны работать с ответами API?

2. Как обрабатывать ошибки сервиса?

3. Как выделить авторизацию и прокидывать собственный токен?

4. Внешний вид библиотеки (ее интерфейс)

Не самые редкие случаи, могут требовать решения и иных вопросов, например: пре/пост обработка ответов API (т.е. использование middleware) или необходимость гибкой работы с неймингом в API, например: преобразовывать user_id в userId и т.д.

Эти вопросы можно решить с помощью моей библиотеки unihttp, давайте рассмотрим, что она умеет и как этим пользоваться. Уверен, у многих уже появилась рабочая необходимость для работы с API мессенджера Max, поэтому выберем его для примера. Начнем с самого базового – API-методы.

Сделаем метод получения профиля бота, и для начала опишем получаемые нами типы:

from dataclasses import dataclass


@dataclass  
class BotCommand:  
    name: str  
    description: str | None = None  
  
  
@dataclass  
class Bot:  
    user_id: int  
    first_name: str  
    is_bot: bool  
    last_activity_time: int  
    last_name: str | None = None  
    username: str | None = None  
    description: str | None = None  
    avatar_url: str | None = None  
    full_avatar_url: str | None = None  
    commands: list[BotCommand] | None = None

Затем нам необходимо создать класс API-запроса:

from unihttp.method import BaseMethod  


@dataclass
class GetMe(BaseMethod[Bot]):  
    __method__ = "GET"  
    __url__ = "/me"

__method__ – HTTP-метод запроса: GET / POST / PATCH и т.д.
__url__ – путь к эндпоинту без учета базового пути (рабо��ает в том числе без / в начале)

И наконец, создадим класс самого клиента:

from niquests import AsyncSession  
from unihttp.bind_method import bind_method  
from unihttp.clients.niquests import NiquestsAsyncClient  
from unihttp.serializers.adaptix import DEFAULT_RETORT


class MaxClient(NiquestsAsyncClient):  
    def __init__(self, token: str) -> None:  
        super().__init__(  
            base_url="https://platform-api.max.ru",  
            request_dumper=DEFAULT_RETORT,  
            response_loader=DEFAULT_RETORT,  
            session=AsyncSession(  
                headers={  
                    "Authorization": token  
                }  
            )  
        )  
  
    get_me = bind_method(GetMe)

Что мы здесь делаем:

1. импортируем основу клиента – NiquestsAsyncClient

2. инициализируем клиент:

   1. base_url – базовый урл API

   2. request_dumper – используется для дампа запроса

   3. response_loader – используется для десериализации ответов

   4. session – сюда мы передаем AsyncSession с хедерами, в которых у нас содержится авторизационный хедер

3. используем bind_method для привязки метода к клиенту

Часть с request_dumper и response_loader мы рассмотрим немного позже.

Итого, инициализация клиента и вызов метода, у нас выглядит следующим образом:

async def main():  
    TOKEN = "Ваш токен от бота"  
  
    client = MaxClient(token=TOKEN)  
    me = await client.get_me()  
    pprint(me)

После запуска этого кода, мы получаем следующее (реальные данные убраны):

Bot(
	user_id=1,
    first_name='имя бта',
    is_bot=True,
    last_activity_time=1,
    last_name=None,
    username='bot',
    description='описание бота',
    avatar_url='https://i.oneme.ru/i?r=12345',
    full_avatar_url='https://i.oneme.ru/i?r=12345',
    commands=None
)

Давайте рассмотрим немного более сложный запрос: закрепление сообщения в групповом чате.

from unihttp.omitted import Omittable, Omitted
from unihttp.markers import Path, Body


@dataclass  
class QueryObject:  
    success: bool  
    message: str | None = None


@dataclass
class PinMessage(BaseMethod[QueryObject]):  
    __url__ = "/chats/{chat_id}/pin"  
    __method__ = "PUT"  
  
    chat_id: Path[int]  
  
    message_id: Body[str]  
    notify: Omittable[bool] = Omitted()

Здесь у нас появились два метода параметров:

1. Path-параметр chat_id

2. Body-параметры: message_id, notify

notify является Omittable параметром, у него нет дефолтного значения в самом методе, и он не будет передан в тело запроса.

Вместо того, чтобы вручную собирать словари для заголовков или заниматься конкатенацией строк для URL, unihttp использует систему специальных маркеров. Это позволяет превратить сигнатуру метода в прозрачное описание HTTP-запроса.

В библиотеке есть 6 маркеров, которые закрывают 100% потребностей:

1. Path – подставляет параметр в __url__

2. Query – автоматически сериализует параметры в query-string

3. Header – добавляет хедеры к запросу

4. Body – параметры, которые должны быть отправлены в JSON-формате в теле запроса

5. Form – параметры, которые используют стандартную URL-кодировку для передачи полей в теле сообщения

6. File – используется для отправки файлов

Обработка ошибок

Давайте теперь сделаем обработку ошибок в клиенте.

В базовых клиентах библиотеки BaseSyncClient и BaseAsyncClient есть два метода: validate_response и handle_error.

validate_response – идея этого метода заключается в том, что у каких-то API могут быть специфичные ответы, что-то вроде:

200 OK
{"ok": false, "error": "some error"}

handle_error – основывается на статус кодах (>= 300)

Вы можете наследовать эти методы в собственных клиентах. На нашем примере, это выглядит так.

Опишем сами классы ошибок:

class MaxException(Exception):  
    pass  
  
  
class InvalidRequestException(MaxException):  
    pass  
  
  
class AuthenticationException(MaxException):  
    pass  
  
  
class ResourceNotFoundException(MaxException):  
    pass  
  
  
class MethodNotAllowedException(MaxException):  
    pass  
  
  
class RateLimitException(MaxException):  
    pass  
  
  
class ServiceUnavailableException(MaxException):  
    pass

Обработка ошибок на уровне клиента выглядит так:

from unihttp.http import HTTPResponse


class MaxClient(NiquestsAsyncClient):
	...

	def handle_error(self, response: HTTPResponse, method: BaseMethod):  
	    error_by_status_codes = {  
	        400: InvalidRequestException,  
	        401: AuthenticationException,  
	        404: ResourceNotFoundException,  
	        405: MethodNotAllowedException,  
	        429: RateLimitException,  
	        503: ServiceUnavailableException,  
	    }  
	    error = error_by_status_codes.get(response.status_code, MaxException)  
	    raise error(response.data)

В сам метод у нас передается response: HTTPResponse, нормализованный класс самого unihttp, который содержит: status_code, headers, data, cookies и raw_response из библиотеки, которую вы используете в качестве бэкэнда для клиента.

Основываясь на этих данных, вы можете выбрасывать любую ошибку, в зависимости от требуемой вам логики.

Middlewares

В библиотеке есть механизм мидлварей, который позволяет вам писать собственные мидлвари, например: логировать все исходящие запросы и ответы на них.

В связи с тем, что в библиотеке поддерживается два вида клиентов: синхронные и асинхронные, есть два вида мидлварей:

1. синхронные (для синхронных клиентов) – используется протокол Middleware

2. асинхронные (для асинхронных клиентов) – используется протокол AsyncMiddleware

Рассмотрим мидлварь простейшего логирования самих запросов и их ответов:

from unihttp.middlewares import AsyncMiddleware, AsyncHandler


class LoggingMiddleware(AsyncMiddleware):  
    async def handle(  
            self, request: HTTPRequest, next_handler: AsyncHandler  
    ) -> HTTPResponse:  
        print("Outgoing request:", request)  
        response = await next_handler(request)  
        print("Outgoing response:", response)  
        return response

Мидлварь выглядит собственно так просто, и никаких сложных вещей здесь нет. В самом клиенте вы их можете указывать следующим образом:

class MaxClient(NiquestsAsyncClient):  
    def __init__(  
            self, token: str,  
            middleware: list[AsyncMiddleware] | None = None,  
    ) -> None:  
        super().__init__(  
            ...,  
            middleware=middleware  
        )
        
client = MaxClient(token=TOKEN, middleware=[LoggingMiddleware()])

Возможные способы вызова методов

Стоит отметить, что в библиотеке существует два способа вызывать какой-либо метод.

bind_method – этот способ мы применили с вами немного ранее. Здесь стоит отметить, что его можно использовать в любом классе, который содержит в себе call_method

call_method – этот способ есть в любом клиенте, и его можно использовать следующим образом:

class MaxClient(NiquestsAsyncClient):  
    ... 
  
    async def get_me(self):
	    return await self.call_method(GetMe())

Это публичный метод, поэтому он доступен к вызову в том числе и просто к инстансу вашего клиента.

json_loads / json_dumps

При инициализации клиента, вы также можете передавать собственные функции для сериализации и десериализации данных из JSON. Например, если вы хотите использовать не стандартную библиотеку json, а ujson, orjson или иные библиотеки.

class MaxClient(NiquestsAsyncClient):  
    def __init__(self, token: str) -> None:  
        super().__init__(  
            ...,  
		    json_dumps=lambda x: orjson.dumps(x).decode(),
		    json_loads=orjson.loads
        )  

Сериализация данных

Мы с вами создавали класс самого клиента, и теперь вернемся к объяснениям про параметры request_dumper и response_loader.

В библиотеке доступно несколько вариантов для (де)сериализации данных:

1. adaptix (рекомендую)

2. pydantic

Рассмотрим их далее:

Adaptix

adaptix – очень мощная и гибкая библиотека для работы с сериализацией, десериализации и конверсии данных.

Что можно делать с помощью адаптикса:

1. менять именование параметров на входе / выходе, например: вам необходимо отправлять userId, правило для работы может выглядеть так: name_mapping(GetUser, map={"user_id": "userId"}. Это правило в том числе может быть применимо и к всем методам: name_mapping(OriginSubclassLSC(BaseMethod), map={"extra_meow": "ExtraMeowParameter"})

2. сериализация данных в нужный формат, возможно вам необходимо приводить дату к формату год-месяц-день, правило будет выглядеть так: dumper(datetime.datetime, lambda x: x.strptime("%Y-%m-%d"))

Отдельно стоит упомянуть, в unihttp реализован for_marker – предикат, который позволяет применять правила к маркерам запроса (Query, Body и т.д.), например, нам необходимо все списки в Query сериализовывать в список значений через запятую. Мы можем сделать это следующим способом:

dumper(
	for_marker(QueryMarker, P[list[str]]),
	lambda x: ",".join(x)
)

У adaptix можно очень гибко настраивать в том числе и объекты применения правил, например:

1. их можно применять к базовым классам

2. к конкретным классам

3. к определенным полям классов

4. к самим типам данным

5. какие-то правила применимы в целом к реторте

Можно также и многое иное, но почитать подробнее об этом можно в самой библиотеке адаптикса ;)

Pydantic

Pydantic — библиотека для Python, предназначенная для валидации и трансформации данных.

Несомненно, что вы все слышали о ней. Использовать pydantic для (де)сериализации данных можно следующим образом.

from unihttp.clients.niquests import NiquestsAsyncClient  
from unihttp.serializers.pydantic import PydanticLoader, PydanticDumper


class MaxClient(NiquestsAsyncClient):  
    def __init__(self, token: str) -> None:  
        super().__init__(  
			# ...
            request_dumper=PydanticLoader,  
            response_loader=PydanticDumper,
        )  

---

На этом обзор unihttp подошел к концу.

Буду рад вашей обратной связи и звездочкам на репозитории ;)

Дополнительные ссылки:

1. Репозиторий – https://github.com/goduni/unihttp

2. Документация adaptix – https://adaptix.readthedocs.io/

3. Фреймворк, который использует unihttp для работы с Max API – https://github.com/K1rL3s/maxo