Реальные библиотеки — asyncio, Twisted, когда нужны готовые tools

В предыдущих уроках мы писали networking code на raw sockets — учили API ОС напрямую. Это даёт глубокое понимание, но в production вы будете использовать библиотеки: asyncio, aiohttp, httpx, Twisted, Tornado, и др. Они скрывают сложность (framing, multiplexing, retry-логика, TLS) и дают высокоуровневые abstractions.

В этом уроке — обзор основных Python библиотек для сетевого программирования. Когда какую использовать, какие они дают преимущества, и где raw sockets всё ещё имеют смысл.

Когда нужны библиотеки vs raw sockets

Raw sockets имеют смысл когда:

1. Образовательные цели — понять, как работает протокол под капотом.
2. Кастомный протокол — свой бинарный формат, не подходящий под стандартные библиотеки.
3. Минимальный binary size — если ваше приложение должно быть в kB (embedded, micropython).
4. Performance критичен — иногда custom epoll-loop быстрее, чем library overhead.

Библиотеки нужны для:

1. HTTP-клиент/сервер. Никогда не пишите HTTP/1.1 на raw sockets для production — слишком много corner cases. Используйте requests, httpx, aiohttp.
2. Любой стандартный протокол — SMTP, FTP, IRC, MQTT, gRPC. Готовые библиотеки протестированы и handle edge cases.
3. TLS — никогда не пишите свою криптографию. ssl модуль или cryptography.
4. Concurrency framework — asyncio, Twisted, gevent. Дают event loops и primitives.

99% сетевого кода в работе — через библиотеки. Raw sockets знание полезно для понимания и редких случаев.

asyncio — стандартная библиотека Python

asyncio — основной async framework Python. Включён в stdlib с Python 3.4 (2014). Эволюционирует, сейчас стабилен и идиоматичен.

Низкоуровневый: streams API

import asyncio

async def main():
    # Подключение
    reader, writer = await asyncio.open_connection('httpbin.org', 80)

    # Отправка запроса
    request = b'GET /get HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n'
    writer.write(request)
    await writer.drain()

    # Чтение ответа
    response = await reader.read()
    print(response.decode()[:500])

    writer.close()
    await writer.wait_closed()

asyncio.run(main())

reader.read(), writer.write() — асинхронные эквиваленты sock.recv(), sock.send(). drain() ждёт, пока write buffer освободится — back-pressure mechanism.

Сервер через high-level API

import asyncio

async def handle_echo(reader, writer):
    while True:
        data = await reader.read(4096)
        if not data:
            break
        writer.write(data)
        await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(handle_echo, '0.0.0.0', 9999)
    async with server:
        await server.serve_forever()

asyncio.run(main())

Намного proще, чем raw epoll. И масштабируется до тысяч-десятков тысяч клиентов.

asyncio.wait_for — timeouts

try:
    data = await asyncio.wait_for(reader.read(), timeout=5.0)
except asyncio.TimeoutError:
    print("Read timed out")

asyncio.gather — параллельные tasks

async def fetch(url):
    # ...
    return data

# Запустить три fetch'а параллельно
results = await asyncio.gather(
    fetch('https://a.com'),
    fetch('https://b.com'),
    fetch('https://c.com'),
)

Все три выполняются одновременно. Каждый — свой async task в event loop. Время = max(a, b, c), а не sum.

httpx — современный HTTP-клиент

requests — классическая Python HTTP-библиотека, sync. Если хотите HTTP в async коде — httpx.

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.github.com/users/torvalds')
        print(response.status_code)
        print(response.json())

asyncio.run(main())

httpx — API почти идентичный requests, плюс async support. Поддерживает HTTP/1.1 и HTTP/2. Connection pooling из коробки.

Параллельные запросы:

async with httpx.AsyncClient() as client:
    urls = ['https://api.github.com/users/torvalds',
            'https://api.github.com/users/gvanrossum',
            'https://api.github.com/users/raymondh']

    tasks = [client.get(url) for url in urls]
    responses = await asyncio.gather(*tasks)

    for r in responses:
        print(r.json()['name'])

Три HTTP-запроса параллельно, общее время = max RTT, а не sum. На I/O bound задачах это огромный win.

aiohttp — HTTP сервер и клиент

aiohttp — более старый и feature-rich пакет. Главное преимущество над httpx — встроенный HTTP-сервер.

Server

from aiohttp import web

async def hello(request):
    return web.Response(text='Hello, world!')

async def user(request):
    user_id = request.match_info['user_id']
    return web.json_response({'user_id': user_id, 'name': 'Linus'})

app = web.Application()
app.router.add_get('/', hello)
app.router.add_get('/users/{user_id}', user)

web.run_app(app, host='0.0.0.0', port=8080)

Production HTTP-server с роутингом, middleware, WebSocket support — в 20 строк.

WebSocket

from aiohttp import web

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            await ws.send_str(f"Echo: {msg.data}")
        elif msg.type == web.WSMsgType.ERROR:
            print(f'WS error: {ws.exception()}')

    return ws

app = web.Application()
app.router.add_get('/ws', websocket_handler)
web.run_app(app)

WebSocket echo-сервер за 15 строк. Без aiohttp пришлось бы вручную реализовывать WebSocket protocol (handshake, frames, masking) — сотни строк сложного кода.

FastAPI — современный web framework

FastAPI — web framework поверх Starlette и Pydantic. Async по умолчанию, type-hints для автодокументации.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    name: str
    email: str

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "Linus"}

@app.post("/users")
async def create_user(user: User):
    return {"user_id": 42, "name": user.name, "email": user.email}

# Запуск: uvicorn main:app --reload

FastAPI автоматически:

• Валидирует входные параметры (Pydantic).
• Генерирует OpenAPI/Swagger docs.
• Использует async для max performance.

Это де-факто стандарт для новых Python API в 2026 году.

Twisted — старый ветеран

Twisted — одна из первых async networking библиотек Python (с 2002 года). Имеет огромный набор протоколов: SMTP, POP3, IRC, NNTP, SSH, и т.д.

from twisted.internet import reactor, protocol

class EchoProtocol(protocol.Protocol):
    def dataReceived(self, data):
        self.transport.write(data)

class EchoFactory(protocol.ServerFactory):
    protocol = EchoProtocol

reactor.listenTCP(9999, EchoFactory())
reactor.run()

Стиль — callback-based, не async/await. Сложнее писать и debug по сравнению с asyncio. В новом коде Twisted используется реже, в основном для legacy и специальных протоколов.

Если у вас есть Twisted-codebase — можно постепенно migrate на asyncio через bridge twisted.internet.asyncioreactor.

Tornado — async без asyncio

Tornado — async web framework, появился в 2009 году в FriendFeed. Использовал свой event loop до Python 3.5. С появлением asyncio Tornado может работать поверх asyncio loop.

import tornado.ioloop
import tornado.web

class HelloHandler(tornado.web.RequestHandler):
    async def get(self):
        self.write("Hello, world!")

app = tornado.web.Application([
    (r"/", HelloHandler),
])

if __name__ == "__main__":
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

Tornado использовался в продакшене многих компаний. Сейчас FastAPI/aiohttp популярнее для нового кода.

Низкоуровневые библиотеки

Для специальных случаев есть:

• pyzmq — ZeroMQ bindings. Messaging library с patterns (pub-sub, req-rep, push-pull). Для distributed systems и IPC.

• paramiko — SSH-клиент и сервер. Если нужно делать SSH-операции из Python.

• scapy — packet manipulation. Создание custom пакетов любого уровня, sniffing. Идеален для network analysis и security testing.

• dnspython — DNS-клиент. Если нужно делать DNS queries из Python (для проверки записей, audit).

• aiosmtpd — async SMTP server. Если нужно принимать почту.

Каждая из них — 5-50K строк кода, реализующая полностью свой протокол. Не пишите своё, если есть готовое.

TLS и aiohttp/httpx

Все современные библиотеки по умолчанию валидируют TLS-сертификаты:

async with httpx.AsyncClient() as client:
    # Это автоматически:
    # - использует HTTPS
    # - валидирует сертификат через certifi (Mozilla CA bundle)
    # - проверяет hostname
    response = await client.get('https://api.github.com/')

Если сертификат невалидный — exception. Чтобы отключить (только для тестов!):

async with httpx.AsyncClient(verify=False) as client:
    # SSL warning будет показан
    ...

Никогда verify=False в production. Только для local testing против self-signed.

Кастомный CA bundle:

async with httpx.AsyncClient(verify='/path/to/ca-bundle.pem') as client:
    ...

Для mTLS (client cert):

async with httpx.AsyncClient(cert=('/path/to/client.crt', '/path/to/client.key')) as client:
    ...

Connection pooling

Производительность HTTP-запросов часто упирается в connection setup — TCP + TLS handshake занимают 100-200ms. Если делаете много запросов на тот же host — переиспользование connection даёт огромный win.

# Плохо -- каждый запрос новый client = новые connections
for url in urls:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)

# Хорошо -- один client, переиспользует connections (HTTP/1.1 keep-alive, HTTP/2 multiplexing)
async with httpx.AsyncClient() as client:
    for url in urls:
        response = await client.get(url)

Для requests: используйте Session — эквивалент. Auto-pooling без явных настроек.

Внутри httpx.AsyncClient — pool из N connections per host (по умолчанию ~10). Можно tune:

limits = httpx.Limits(max_connections=100, max_keepalive_connections=20)
async with httpx.AsyncClient(limits=limits) as client:
    ...

Когда писать свой networking code

Случаи, когда библиотек недостаточно:

1. Бинарный кастомный протокол. Самописный протокол для game server, IoT device communication, propriety RPC. Например, ваш sensor шлёт packed binary data специфической структуры — библиотеки не помогут, нужно struct + raw socket.

2. Производительность. Если профайлер показывает, что библиотека — bottleneck, иногда custom code эффективнее. Очень редкий случай.

3. Embedded constraints. MicroPython, embedded Linux с ограничениями — легче минимальный socket-код.

4. Protocol research/reverse engineering. Хотите понять, как работает чужой protocol, посмотреть пакеты, может фуззить — raw sockets + scapy.

В 99% работы junior data engineer’а вы используете высокоуровневые библиотеки. Это нормально и правильно. Знание raw sockets даёт understanding, но писать с нуля — избыточная сложность.

Реальный пример: production-grade HTTP-клиент

Типичный pattern для скрипта, который обращается к внешним APIs:

import asyncio
import httpx
from typing import Optional

async def fetch_with_retry(
    client: httpx.AsyncClient,
    url: str,
    max_retries: int = 3,
    backoff_factor: float = 2.0,
) -> Optional[dict]:
    """
    Делаем GET с retry на 5xx и timeouts.
    """
    for attempt in range(max_retries):
        try:
            response = await client.get(url, timeout=10.0)
            if response.status_code == 200:
                return response.json()
            elif 500 <= response.status_code < 600:
                # Серверная ошибка -- retry
                await asyncio.sleep(backoff_factor ** attempt)
                continue
            else:
                # 4xx -- клиентская ошибка, не ретраим
                response.raise_for_status()
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(backoff_factor ** attempt)

    return None

async def main():
    limits = httpx.Limits(max_connections=10)
    async with httpx.AsyncClient(limits=limits) as client:
        users = ['torvalds', 'gvanrossum', 'mojombo']

        tasks = [
            fetch_with_retry(client, f'https://api.github.com/users/{u}')
            for u in users
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for user, result in zip(users, results):
            if isinstance(result, Exception):
                print(f"{user}: error {result}")
            else:
                print(f"{user}: {result['name']}")

asyncio.run(main())

В 40 строк — robust async HTTP-клиент с retry, timeouts, connection pooling, parallelism, error handling. Реализовать всё это на raw sockets — сотни строк сложного кода с edge cases.

Попробуй сам

# 1. Установить современные libs
pip install httpx aiohttp fastapi uvicorn

# 2. Простой async HTTP-клиент
python3 -c "
import asyncio, httpx
async def main():
    async with httpx.AsyncClient() as client:
        r = await client.get('https://api.github.com/users/torvalds')
        print(r.json()['name'])
asyncio.run(main())
"

# 3. Параллельные запросы -- сравнить timing с sync
python3 -c "
import time
import httpx
import asyncio

urls = ['https://api.github.com/users/torvalds',
        'https://api.github.com/users/gvanrossum',
        'https://api.github.com/users/mojombo']

# Sync
start = time.time()
for url in urls:
    httpx.get(url)
print(f'Sync: {time.time() - start:.2f}s')

# Async
async def async_test():
    async with httpx.AsyncClient() as client:
        await asyncio.gather(*[client.get(u) for u in urls])
start = time.time()
asyncio.run(async_test())
print(f'Async: {time.time() - start:.2f}s')
"
# Видим: async в 3-5 раз быстрее на 3 URLs

# 4. Простой FastAPI server
python3 -c "
from fastapi import FastAPI
app = FastAPI()

@app.get('/')
def root():
    return {'hello': 'world'}

@app.get('/users/{user_id}')
def get_user(user_id: int):
    return {'user_id': user_id}
" > server.py

# Запустить
# uvicorn server:app --reload
# Открыть http://localhost:8000/docs -- автоматический Swagger UI

# 5. aiohttp WebSocket echo
# Возьмите код из урока, сохраните в ws_server.py
# В другом окне: pip install websockets
# python3 -c "
# import asyncio
# import websockets
# async def echo():
#     async with websockets.connect('ws://localhost:8080/ws') as ws:
#         await ws.send('Hello!')
#         response = await ws.recv()
#         print(response)
# asyncio.run(echo())
# "

# 6. Сравнить producitivity: написать REST API на flask vs FastAPI
# Flask -- imperative, sync, ручная валидация
# FastAPI -- declarative, async, auto-validation

Что вы должны вынести

1. В production используются библиотеки — raw sockets только для специальных случаев.
2. asyncio — async framework stdlib. Streams API (high-level) и transport API (low-level).
3. httpx — современный HTTP-клиент (sync + async). FastAPI / aiohttp — веб-фреймворки.
4. Connection pooling — через переиспользование Client / Session. Без него производительность в 10x хуже.
5. TLS-валидация по умолчанию — никогда не отключать в production.
6. Параллельные операции через asyncio.gather — огромный win для I/O bound.
7. Twisted / Tornado — старые, но всё ещё используются. Новый код обычно asyncio-based.

Этот урок завершает модуль о sockets. Дальше идёт балансировка и security.