Декларативный асинхронный и синхронный HTTP и WebSocket клиент для Python на базе HTTPX и Pydantic.
@Route("https://api.example.com/v1")
class UserClient(WApiClient):
@GET("/users/{user_id}")
def get_user(self, user_id: int) -> User: pass
@WebSocket("/stream/{room}")
async def stream_events(self, room: str): passБиблиотека what-wapi предоставляет декларативный интерфейс для создания типизированных HTTP и WebSocket клиентов в Python. Архитектурная концепция вдохновлена библиотеками Retrofit и Feign: сетевые контракты объявляются в виде сигнатур методов с декораторами маршрутов, а вся низкоуровневая рутина по формированию запросов, подстановке параметров, сериализации и десериализации данных выполняется автоматически.
Библиотека построена на базе современного сетевого стека httpx, pydantic v2 и websockets. Она обеспечивает прозрачную работу в синхронном и асинхронном режимах, изолирует состояние инстансов без мутации глобальных атрибутов классов и гарантирует надежное управление жизненным циклом соединений.
-
Декларативное описание контрактов
Маршруты, HTTP-глаголы и WebSocket-каналы описываются декораторами@Route,@GET,@POST,@PUT,@DELETE,@PATCH,@HEAD,@OPTIONSи@WebSocket. -
Интроспекция аргументов и маппинг параметров
Позиционные и именованные аргументы методов автоматически связываются черезinspect.signatureс шаблонами URL пути{param}, query-параметрами или телом запроса. -
Строгая валидация и типизация
Встроенная интеграция с Pydantic v2 и стандартными dataclass обеспечивает валидацию входящих и исходящих данных без ручного парсинга словарей. -
Единый движок HTTPX
Поддержка пула соединений, Keep-Alive, HTTP/2, единых политик таймаутов и корректного чтения тела ответов в асинхронном контексте. -
Полноценная поддержка WebSocket
Декларативное объявление постоянных соединений с автоконвертацией протоколов, контекстным менеджером и типизированной отправкой/приемом сообщений.
wapi/
├── client.py Базовый клиент WApiClient с пулом и контекстным менеджером
├── executor.py Движок выполнения HTTP-запросов и маппинга сигнатур
├── http_methods.py Фабрика декораторов HTTP-методов (GET, POST, PUT, etc.)
├── routing.py Маршрутизатор Route и безопасная иерархия сервисов
├── websocket.py Декларативный декоратор WebSocket и обертка соединения
├── static/ Константы параметров, HTTP-методов и TypeVar
└── utils/
├── serializer.py Сериализация и десериализация Pydantic и dataclass
└── url.py RFC-совместимое слияние путей и парсинг шаблонов
Установка через пакетный менеджер pip:
pip install what-wapiДобавление зависимости в pyproject.toml:
[project]
dependencies = [
"what-wapi>=0.1.0",
]Импорт поддерживается как по имени what_wapi, так и через лаконичный алиас wapi:
from what_wapi import Route, GET, POST, WebSocket, WApiClient
# или
from wapi import Route, GET, POST, WebSocket, WApiClientМетоды объявляются с помощью декораторов соответствующих HTTP-глаголов. В качестве второго аргумента декоратора можно передать целевой тип десериализации, либо указать его в возвращаемом типе аннотации Python.
from typing import List
from pydantic import BaseModel
from what_wapi import Route, GET, POST, PUT, DELETE, WApiClient
class Article(BaseModel):
id: int
title: str
content: str
class CreateArticleRequest(BaseModel):
title: str
content: str
@Route("https://api.example.com")
class BlogService(WApiClient):
@GET("/articles", List[Article])
def get_articles(self) -> List[Article]:
pass
@GET("/articles/{article_id}", Article)
def get_article(self, article_id: int) -> Article:
pass
@POST("/articles", Article)
def create_article(self, body: CreateArticleRequest) -> Article:
pass
@DELETE("/articles/{article_id}")
def delete_article(self, article_id: int):
pass
client = BlogService()
# Запрос списка
articles = client.get_articles()
# Запрос с позиционной подстановкой ID
article = client.get_article(10)
# Создание ресурса
new_article = client.create_article(
CreateArticleRequest(title="Архитектура API", content="Текст статьи")
)
# Удаление (возвращает сырой httpx.Response при отсутствии типа)
response = client.delete_article(10)
assert response.status_code == 204Библиотека анализирует аргументы функции: параметры, совпадающие с плейсхолдерами в строке URL, подставляются в путь; остальные скалярные аргументы конвертируются в query-параметры. Исходные словари пользователя при этом не модифицируются.
@Route("https://api.example.com")
class CatalogService(WApiClient):
@GET("/catalog/{category}")
def search_items(
self,
category: str, # Подставляется в URL: /catalog/books
query: str, # Query-параметр: ?query=python
page: int = 1, # Query-параметр: &page=1
in_stock: bool = True # Query-параметр: &in_stock=true
) -> List[Item]:
pass
catalog = CatalogService()
items = catalog.search_items("books", query="python", page=2, in_stock=True)
# Итоговый URL: https://api.example.com/catalog/books?query=python&page=2&in_stock=trueДля перехода к асинхронным неблокирующим вызовам достаточно объявить метод как async def. Движок использует httpx.AsyncClient и полностью сохраняет тело ответа для последующего чтения.
import asyncio
from what_wapi import Route, GET, WApiClient
@Route("https://api.example.com")
class AsyncClient(WApiClient):
@GET("/data/{item_id}")
async def fetch_data(self, item_id: int) -> Item:
pass
async def main():
async with AsyncClient() as client:
item = await client.fetch_data(42)
print(item)
asyncio.run(main())Сервисы можно структурировать и объединять во вложенные модули. В отличие от наивных реализаций, пути объединяются динамически в рантайме без перезаписи глобального состояния классов, что исключает взаимное влияние инстансов.
@Route("/billing")
class BillingService:
@GET("/invoices/{invoice_id}")
def get_invoice(self, invoice_id: str) -> Invoice:
pass
@Route("/users")
class UserService:
@GET("/{user_id}")
def get_profile(self, user_id: int) -> Profile:
pass
@Route("https://api.company.internal/v2")
class GatewayClient(WApiClient):
users = UserService()
billing = BillingService()
gateway = GatewayClient()
# Вызовы по вложенным путям:
# GET https://api.company.internal/v2/users/100
profile = gateway.users.get_profile(100)
# GET https://api.company.internal/v2/billing/invoices/inv_99
invoice = gateway.billing.get_invoice("inv_99")Декоратор @WebSocket (или его алиас @WS) обеспечивает поддержку постоянных двусторонних соединений. Если базовый URL использует схемы http:// или https://, библиотека автоматически преобразует их в ws:// или wss://.
import asyncio
from pydantic import BaseModel
from what_wapi import Route, WebSocket, WApiClient
class ChatMessage(BaseModel):
author: str
text: str
@Route("wss://stream.example.com")
class RealtimeClient(WApiClient):
@WebSocket("/rooms/{room_id}/live")
async def connect_room(self, room_id: str):
pass
async def run_chat():
client = RealtimeClient()
# Подключение через асинхронный контекстный менеджер
async with client.connect_room("engineering") as ws:
# Отправка текстовых данных
await ws.send_text("Подключен к каналу")
# Отправка типизированной Pydantic модели (автоматический JSON)
msg = ChatMessage(author="Алексей", text="Приветствую")
await ws.send_json(msg)
# Прием и автоматическая валидация входящего JSON в модель
incoming = await ws.recv_json(ChatMessage)
print(f"[{incoming.author}]: {incoming.text}")
# Потоковое чтение входящих сообщений
async for raw_message in ws:
print("Новое событие:", raw_message)
break
asyncio.run(run_chat())Клиент поддерживает централизованную настройку заголовков, кук и таймаутов, а также их точечное переопределение в конкретном запросе.
# Базовая конфигурация на уровне инстанса
client = BlogService(
base_url="https://staging.example.com",
headers={"Authorization": "Bearer SECRETE_TOKEN"},
timeout=15.0 # Дефолтный таймаут в секундах
)
# Переопределение параметров на уровне отдельного вызова
article = client.get_article(
1,
headers={"X-Debug-Trace": "true"},
timeout=5.0
)Запуск тестового набора:
# Клонирование репозитория
git clone https://github.com/whatrushki/wapi.git
cd wapi
# Запуск тестов
pytest -vСборка дистрибутива:
python -m build- Запросы функционала и сообщения об ошибках принимаются через GitHub Issues.
- Официальные каналы коммуникации и сообщество разработчиков: Telegram.