Модуль 7 · Продвинутые техники

Урок 7.3 — Кастомные сессии

Узнаём, как создавать собственные HTTP-сессии для aiosend на базе httpx, requests и других библиотек, настраивать таймауты и прокси.

По умолчанию aiosend использует AiohttpSession на базе aiohttp. Но ты можешь создать свою сессию, используя любую HTTP-библиотеку: httpx, requests (через threading), curl_cffi и другие. Для этого нужно унаследоваться от BaseSession и реализовать метод request().

🔌

AiohttpSession — стандартная сессия

AiohttpSession используется по умолчанию. Она построена на базе aiohttp.ClientSession и подходит для 99% проектов. Рассмотрим её внутреннее устройство.

Python · Внутреннее устройство AiohttpSession

import ssl
import certifi
from aiohttp import ClientSession, ClientTimeout, TCPConnector

class AiohttpSession(BaseSession):
    """Сессия на базе aiohttp."""

    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        self._session: ClientSession | None = None

    async def request(self, token, client, method):
        ssl_context = ssl.create_default_context(cafile=certifi.where())
        self._session = ClientSession(
            timeout=ClientTimeout(self.timeout),
            connector=TCPConnector(ssl_context=ssl_context),
        )
        async with self._session as session:
            resp = await session.post(
                url=self.network.url(method),
                data=method.model_dump_json(exclude_none=True),
                headers={
                    "Crypto-Pay-API-Token": token,
                    "Content-Type": "application/json",
                    "User-Agent": f"Python/3.x aiohttp/x aiosend/3.x",
                },
            )
            response = self._check_response(client, method, await resp.text())
        return response.result

💡 Почему AiohttpSession по умолчанию?

aiohttp — это самая популярная асинхронная HTTP-библиотека в Python. Она быстрая, хорошо поддерживается, имеет встроенную поддержку SSL, таймаутов и пула соединений. Для большинства проектов её более чем достаточно.

📐

BaseSession — абстрактный базовый класс

BaseSession — это абстрактный базовый класс (ABC), от которого должны наследоваться все сессии. Он определяет интерфейс и предоставляет вспомогательные методы для обработки ответов API.

Метод / Атрибут	Описание
`__init__(network, timeout)`	Сохраняет network и timeout.
`request(token, client, method)`	Абстрактный метод — ты должен его реализовать.
`_check_response(client, method, text)`	Парсит ответ API и проверяет на ошибки.
`network`	Объект Network (c полями name, base).

Python · Минимальная реализация BaseSession

from aiosend.client.session.base import BaseSession
from aiosend.client.network import Network

class MyCustomSession(BaseSession):
    """Моя кастомная сессия."""

    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        # Здесь инициализируем свой HTTP-клиент

    async def request(self, token, client, method):
        """
        Отправляет POST-запрос к API.
        
        Аргументы:
            token — API-токен
            client — ссылка на клиент (необязательно)
            method — объект метода API (например, create_invoice)
        """
        # 1. Формируем URL
        url = self.network.url(method)
        
        # 2. Формируем заголовки
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        
        # 3. Сериализуем тело запроса
        data = method.model_dump_json(exclude_none=True)
        
        # 4. Отправляем запрос (твоя реализация)
        response_text = await self._send_post(url, headers, data)
        
        # 5. Обрабатываем ответ через встроенный метод
        return self._check_response(client, method, response_text)

    async def _send_post(self, url, headers, data):
        # Твоя логика HTTP-запроса
        raise NotImplementedError

⚡

Кастомная сессия на httpx

httpx — современная альтернатива aiohttp. Она имеет схожий API, поддерживает HTTP/2, таймауты, прокси и многое другое.

Python · Сессия на httpx

from aiosend import CryptoPay
from aiosend.client.session.base import BaseSession
from aiosend.client.network import Network
import httpx

class HttpxSession(BaseSession):
    """Сессия на базе httpx."""

    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(timeout),
        )

    async def request(self, token, client, method):
        url = self.network.url(method)
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        data = method.model_dump_json(exclude_none=True)

        resp = await self._client.post(url, content=data, headers=headers)
        return self._check_response(client, method, resp.text)

# Использование:
cp = CryptoPay(
    token="1234:TOKEN",
    session=HttpxSession,  # передаём КЛАСС, не объект!
    timeout=60,
)

🔍 Важно: session — это класс, а не объект

Параметр session принимает класс (type), а не экземпляр. aiosend внутри вызывает session(network, timeout), чтобы создать экземпляр, передав ему текущую сеть и таймаут. Это сделано для того, чтобы сессия могла использовать network и timeout из конструктора CryptoPay.

Настройка таймаута для httpx

Python · HttpxSession с тонкой настройкой таймаута

class HttpxSessionWithTimeout(BaseSession):
    """Сессия на httpx с раздельными таймаутами."""

    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        # httpx позволяет раздельно настроить:
        # connect, read, write, pool — таймауты
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=10.0,    # подключение
                read=timeout,     # чтение ответа
                write=10.0,       # запись запроса
                pool=30.0,        # ожидание в пуле
            ),
        )

    async def request(self, token, client, method):
        # ... та же реализация, что выше
        url = self.network.url(method)
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        data = method.model_dump_json(exclude_none=True)
        resp = await self._client.post(url, content=data, headers=headers)
        return self._check_response(client, method, resp.text)

🐌

Кастомная сессия на requests (синхронная)

requests — синхронная библиотека. Чтобы использовать её в асинхронном контексте, нужно запускать запрос в отдельном потоке через run_in_executor или asyncio.to_thread.

Python · Сессия на requests

import asyncio
import requests
from aiosend.client.session.base import BaseSession
from aiosend.client.network import Network

class RequestsSession(BaseSession):
    """Сессия на базе requests (синхронная, через поток)."""

    def __init__(self, network: Network, timeout: float = 300):
        super().__init__(network, timeout)
        self._session = requests.Session()
        self._session.headers.update({
            "Content-Type": "application/json",
        })

    async def request(self, token, client, method):
        url = self.network.url(method)
        headers = {"Crypto-Pay-API-Token": token}
        data = method.model_dump_json(exclude_none=True)

        # Запускаем синхронный запрос в потоке
        resp = await asyncio.to_thread(
            self._session.post,
            url,
            data=data,
            headers=headers,
            timeout=self.timeout,
        )
        return self._check_response(client, method, resp.text)

# Использование:
cp = CryptoPay(
    token="1234:TOKEN",
    session=RequestsSession,
)

⚠️ Когда использовать requests?

requests — хороший выбор, если у тебя уже есть код на requests и ты не хочешь добавлять новую зависимость. Однако для новых проектов рекомендуется httpx или aiohttp — они асинхронные и быстрее.

🔒

Прокси для сессий

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

Прокси через AiohttpSession (расширение)

Стандартная AiohttpSession не поддерживает прокси напрямую. Но ты можешь создать подкласс и добавить поддержку:

Python · AiohttpSession с прокси

from aiosend.client.session.aiohttp import AiohttpSession

class ProxiedAiohttpSession(AiohttpSession):
    """Aiohttp-сессия с поддержкой прокси."""

    def __init__(self, network, timeout=300, proxy=None):
        super().__init__(network, timeout)
        self._proxy = proxy

    async def request(self, token, client, method):
        import ssl
        import certifi
        from aiohttp import ClientSession, ClientTimeout, TCPConnector

        ssl_context = ssl.create_default_context(cafile=certifi.where())
        async with ClientSession(
            timeout=ClientTimeout(self.timeout),
            connector=TCPConnector(ssl_context=ssl_context),
        ) as session:
            resp = await session.post(
                url=self.network.url(method),
                data=method.model_dump_json(exclude_none=True),
                headers={
                    "Crypto-Pay-API-Token": token,
                    "Content-Type": "application/json",
                },
                proxy=self._proxy,  # используем прокси
            )
            return self._check_response(
                client, method, await resp.text()
            )

# Использование:
cp = CryptoPay(
    token="1234:TOKEN",
    session=ProxiedAiohttpSession,
    proxy="http://user:pass@proxy.example.com:8080",
)

Прокси через httpx

httpx имеет встроенную поддержку прокси:

Python · HttpxSession с прокси

import httpx
from aiosend.client.session.base import BaseSession

class HttpxProxySession(BaseSession):
    """httpx-сессия с прокси."""

    def __init__(self, network, timeout=300, proxy_url=None):
        super().__init__(network, timeout)
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(timeout),
            proxies=proxy_url,  # httpx поддерживает прокси
        )

    async def request(self, token, client, method):
        url = self.network.url(method)
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        data = method.model_dump_json(exclude_none=True)
        resp = await self._client.post(url, content=data, headers=headers)
        return self._check_response(client, method, resp.text)

# Использование:
cp = CryptoPay(
    token="1234:TOKEN",
    session=HttpxProxySession,
    proxy_url="http://user:pass@proxy:8080",
)

💡 Альтернатива: кастомная сеть

Вместо прокси-сессии можно использовать кастомный Network с URL прокси-сервера. Если твой прокси просто перенаправляет запросы к API, то достаточно создать Network с базовым URL прокси, не меняя сессию:

Python · Network как альтернатива прокси

from aiosend.client.network import Network

PROXY_NET = Network(
    name="PROXY",
    base="https://my-proxy.com/cryptobot/api/{method}",
)
cp = CryptoPay(token="TOKEN", network=PROXY_NET)

⏱️

Настройка таймаутов

Таймаут определяет максимальное время ожидания ответа от сервера API. По умолчанию — 300 секунд. В кастомной сессии ты можешь настроить его более гибко.

Вид таймаута	Описание	Рекомендация
connect	Время на установку соединения	10-30 секунд
read	Время на чтение ответа	30-300 секунд
pool	Время ожидания в пуле соединений	10-30 секунд

Python · Раздельные таймауты в кастомной сессии

class CustomTimeoutSession(BaseSession):
    """Сессия с раздельными таймаутами."""

    def __init__(self, network, timeout=300):
        super().__init__(network, timeout)
        # Используем timeout из конструктора как read timeout
        # connect, write, pool — фиксированные
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=15.0,
                read=timeout,     # наследуется от CryptoPay(timeout=...)
                write=15.0,
                pool=20.0,
            ),
        )

    async def request(self, token, client, method):
        url = self.network.url(method)
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        data = method.model_dump_json(exclude_none=True)
        resp = await self._client.post(url, content=data, headers=headers)
        return self._check_response(client, method, resp.text)

🚀

Полный пример: выбор сессии

Python · Сравнение сессий

import asyncio
from aiosend import CryptoPay

# 1. Стандартная AiohttpSession (по умолчанию)
cp_aiohttp = CryptoPay(token="1234:TOKEN")

# 2. HttpxSession (кастомная)
cp_httpx = CryptoPay(
    token="1234:TOKEN",
    session=HttpxSession,
    timeout=60,
)

# 3. RequestsSession (через потоки)
cp_requests = CryptoPay(
    token="1234:TOKEN",
    session=RequestsSession,
)

# 4. Сессия с прокси
cp_proxied = CryptoPay(
    token="1234:TOKEN",
    session=ProxiedAiohttpSession,
    proxy="http://user:pass@proxy:8080",
)

async def check_all():
    for name, cp in [
        ("Aiohttp", cp_aiohttp),
        ("Httpx", cp_httpx),
        ("Requests", cp_requests),
        ("Proxied", cp_proxied),
    ]:
        me = await cp.get_me()
        print(f"{name}: {me.name} (через {cp.session.__class__.__name__})")

asyncio.run(check_all())

📌

Что важно запомнить

1️⃣

BaseSession — абстрактный базовый класс. Все сессии должны наследоваться от него и реализовать метод request().

2️⃣

session — это класс, а не объект. В конструктор CryptoPay передаётся класс (type), aiosend сам создаёт экземпляр, передавая network и timeout.

3️⃣

Можно использовать httpx, requests, любую другую библиотеку. Для синхронных библиотек используй asyncio.to_thread().

4️⃣

Прокси поддерживаются. Через кастомную сессию или через кастомный Network.

5️⃣

Не забывай про _check_response(). Всегда используй встроенный метод для обработки ответа — он парсит JSON, проверяет ошибки и возвращает результат.

🎯

Практическое задание

Задание: Создай кастомную сессию на httpx

Напиши и протестируй кастомную сессию на httpx:

Создай класс MyHttpxSession, наследующий от BaseSession
Реализуй __init__ с созданием httpx.AsyncClient
Реализуй request() с отправкой POST и обработкой через _check_response
Создай CryptoPay с этой сессией
Вызови get_me() для проверки

Подсказка:

Python · Шаблон решения

import asyncio
import httpx
from aiosend import CryptoPay
from aiosend.client.session.base import BaseSession

class MyHttpxSession(BaseSession):
    def __init__(self, network, timeout=300):
        super().__init__(network, timeout)
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(timeout),
        )

    async def request(self, token, client, method):
        url = self.network.url(method)
        headers = {
            "Crypto-Pay-API-Token": token,
            "Content-Type": "application/json",
        }
        data = method.model_dump_json(exclude_none=True)
        resp = await self._client.post(url, content=data, headers=headers)
        return self._check_response(client, method, resp.text)

async def main():
    cp = CryptoPay(
        token="YOUR_TOKEN",
        session=MyHttpxSession,
    )
    me = await cp.get_me()
    print(f"Успех: {me.name}")

asyncio.run(main())

Урок 7.3: Кастомные сессии

5 вопросов