Когда в 1991 году Гвидо ван Россум представил миру Python, никто не мог предсказать, какое место через несколько десятилетий этот язык займет в веб-разработке, Data Science и Machine Learning. Сейчас Python продолжает развиваться: с новым поколением инструментов в прошлое уходят традиционные ограничения — производительность, GIL и сложность параллельных вычислений. 

Привет, Хабр! С вами Леша Жиряков, я руковожу бэкенд-направлением витрины KION, возглавляю гильдию по Python и пишу для блога MWS на Хабре. Я каждый день сталкиваюсь с вызовами высоконагруженных систем и сформировался пул инструментов, которые помогают решать критические проблемы современной разработки — от обработки данных с Polars до управления зависимостями с UV. 

В этом материале я сделаю обзор Python-библиотек, с которыми можно создавать системы, сравнимые по производительности с Go и Rust. 

Содержание

• FastAPI — современный асинхронный веб-фреймворк

• Litestar: мощный веб-фреймворк, который догоняет лидеров

• Polars в деле: быстрый анализ данных без боли

• HTTPX — HTTP-клиент от создателей FastAPI

• Dask: параллельные вычисления без компромиссов

• Pydantic V2: Революция в валидации данных Python

• Ruff: новый стандарт линтинга для Python-разработчиков

• Python UV: революция в управлении пакетами Python

FastAPI — современный асинхронный веб-фреймворк

FastAPI — это крутой и быстрый веб-фреймворк на Python 3.7+ для создания API. В последние годы он набрал популярность: причина этого успеха кроется в сочетании скорости разработки, производительности и автоматической валидации данных. У FastAPI есть несколько выигрышных характеристик.

Скорость

Фреймворк супербыстрый. Он работает на ASGI-сервере Starlette и проверяет данные через Pydantic, поэтому FastAPI такой же быстрый, как NodeJS и Go. Получается, это один из самых резвых Python-фреймворков по рынку.

Интуитивно понятный синтаксис и типизация

Вот как просто выглядит код в FastAPI: 

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

@app.post("/items/")
def create_item(item: Item):
    return item

Вы задаете модель данных через Pydantic и используете декораторы для настройки маршрутов. Python-аннотации типов здесь помогают автоматически проверять запросы.

Генерация документации

Это еще одна фишка FastAPI: запустил приложение, зашел по адресу /docs — и готово:

# После запуска приложения
# http://127.0.0.1:8000/docs

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

Мощная проверка данных 

FastAPI следит за всем: от размера текста до сложных правил бизнеса. И если что-то не так, выдает понятные ошибки в JSON.

from fastapi import FastAPI, Path, Query
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str = Field(..., min_length=3, max_length=50)
    price: float = Field(..., gt=0)
    tags: list[str] = []

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(..., ge=1, description="ID товара"),
    q: str = Query(None, max_length=50, description="Поисковый запрос")
):
    return {"item_id": item_id, "q": q}

Встроенная поддержка асинхронности

FastAPI поддерживает async/await, что важно для высоконагруженных приложений: 

@app.get("/async-items/")
async def read_items():
    items = await database.fetch_all("SELECT * FROM items")
    return items

Безопасность

Есть встроенные инструменты для разных схем аутентификации, например OAuth2 с JWT:

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    user = get_user_from_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

Благодаря новейшим возможностям Python, таким как аннотации типов, асинхронность и проверка через Pydantic, с Fast API можно делать надежные API с хорошей документацией и минимумом кода. Если вы создаете микросервисы, бэкенд для SPA или мобильных приложений или просто хотите современный REST API, FastAPI вам поможет. В KION мы с его помощью делаем персональные витрины и применяем бизнес-правила. Подробнее об этом можно почитать здесь.

И вроде бы все хорошо, но есть нюанс. Как я говорил в посте FastAPI vs Litestar, изменения, которые касаются глубокой функциональности FastAPI, вносит только основатель фреймворка — Себастьян Рамирес. 

Litestar: мощный веб-фреймворк, который догоняет лидеров

Это новичок в мире Python. Litestar создан на основе ASGI: с продвинутым функциональностью и при этом не теряет в скорости. В нем есть несколько особенностей.

Компонентный подход

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

from litestar import Litestar, Controller, get

class UserController(Controller):
    path = "/users"
    
    @get("/")
    async def get_users(self) -> list[dict]:
        return [{"id": 1, "name": "Алексей"}, {"id": 2, "name": "Мария"}]
    
    @get("/{user_id:int}")
    async def get_user(self, user_id: int) -> dict:
        return {"id": user_id, "name": "Пользователь " + str(user_id)}

app = Litestar(route_handlers=[UserController])

Встроенная проверка данных

Никаких дополнительных библиотек добавлять не нужно:

from dataclasses import dataclass
from litestar import Litestar, post
from litestar.dto import DTOData

@dataclass
class UserCreate:
    name: str
    email: str
    age: int

@post("/users", dto=UserCreate)
async def create_user(data: DTOData[UserCreate]) -> dict:
    user = data.create_instance()
    # В реальном приложении здесь был бы код для сохранения в БД
    return {"id": 1, "name": user.name, "email": user.email, "age": user.age}

app = Litestar(route_handlers=[create_user])

Мощная и гибкая система для работы с зависимостями:

from litestar import Litestar, get, Dependency
from litestar.di import Provide

async def get_username(request) -> str:
    return request.headers.get("x-username", "гость")

@get("/welcome", dependencies={"username": Provide(get_username)})
async def welcome(username: str) -> dict:
    return {"message": f"Добро пожаловать, {username}!"}

app = Litestar(route_handlers=[welcome])

Встроенная система кеширования:

from litestar import Litestar, get
from litestar.cache import Cache

@get("/expensive-operation", cache=Cache(expiration=60))
async def expensive_operation() -> dict:
    # Имитация дорогостоящей операции
    import time
    time.sleep(2)
    return {"result": "Данные, которые дорого вычислять"}

app = Litestar(route_handlers=[expensive_operation])

WebSocket-соединения с высокой скоростью работы

 У Litestar легкий и понятный API для работы с WebSocket:

from litestar import Litestar, WebSocket
from litestar.handlers import websocket

@websocket("/ws")
async def websocket_endpoint(socket: WebSocket) -> None:
    await socket.accept()
    while True:
        data = await socket.receive_text()
        await socket.send_text(f"Вы отправили: {data}")

app = Litestar(route_handlers=[websocket_endpoint])

Сравнение производительности Litestar с FastAPI

Начнем со времени обработки запросов:

# Приложение на Litestar
from litestar import Litestar, get

@get("/")
async def litestar_handler() -> dict:
    return {"message": "Hello World"}

app_litestar = Litestar(route_handlers=[litestar_handler])

# Приложение на FastAPI
from fastapi import FastAPI

app_fastapi = FastAPI()

@app_fastapi.get("/")
async def fastapi_handler():
    return {"message": "Hello World"}

Результаты бенчмарка (запросов в секунду):

• Litestar: ~15000 RPS

• FastAPI: ~10000 RPS

Использование памяти. Тест нагрузки с 1000 одновременных подключений:

• Litestar: ~65 МБ

• FastAPI: ~90 МБ

Время запуска приложения:

# Измерение времени запуска
import time

start = time.time()
from litestar import Litestar
app = Litestar(route_handlers=[])
end = time.time()
print(f"Litestar startup time: {end - start:.4f} seconds")

start = time.time()
from fastapi import FastAPI
app = FastAPI()
end = time.time()
print(f"FastAPI startup time: {end - start:.4f} seconds")

Результаты:

• Litestar startup time: 0.1245 seconds

• FastAPI startup time: 0.1890 seconds

Комплексный пример с валидацией данных:

# Litestar
from litestar import Litestar, post
from dataclasses import dataclass

@dataclass
class Item:
    name: str
    price: float
    
@post("/items")
async def create_item_litestar(data: Item) -> dict:
    return {"name": data.name, "price": data.price, "tax": data.price  0.2}

# FastAPI
from fastapi import FastAPI
from pydantic import BaseModel

class ItemModel(BaseModel):
    name: str
    price: float

app = FastAPI()

@app.post("/items")
async def create_item_fastapi(item: ItemModel):
    return {"name": item.name, "price": item.price, "tax": item.price  0.2}

Результаты бенчмарка при 10000 запросов с валидацией:

• Litestar: 0.85 сек (11764 RPS)

• FastAPI: 1.27 сек (7874 RPS)

Ключевые преимущества Litestar перед FastAPI:

• более высокая производительность — в среднем на 30–50% быстрее;

• меньшее потребление памяти;

• встроенная система кеширования;

• более гибкая компонентная архитектура;

• отличная поддержка типизации и интеграция с Python-экосистемой.

При этом Litestar сохраняет простоту использования и у него отличная документация. Это хороший выбор как для новых проектов, так и для миграции с существующих фреймворков.

Polars в деле: быстрый анализ данных без боли

Долгое время анализ данных на Python опирался в основном на Pandas, но сейчас все чаще внимание переключается на Polars — более производительную и современную библиотеку, написанную на Rust, но с оберткой для Python. Она создана, чтобы обрабатывать большие объемы данных, почти не расходуя память.

Плюсы Polars

Производительность Polars заметно выше, чем Pandas, особенно на больших наборах данных:

import pandas as pd
import polars as pl
import time
import numpy as np

# Создаем большой набор данных
n = 10_000_000
data = {
    "id": np.arange(n),
    "value1": np.random.rand(n),
    "value2": np.random.rand(n),
    "category": np.random.choice(["A", "B", "C", "D"], n)
}

# Pandas
df_pd = pd.DataFrame(data)
start = time.time()
result_pd = df_pd.groupby("category")["value1"].mean()
pd_time = time.time() - start
print(f"Pandas время выполнения: {pd_time:.4f} сек")

# Polars
df_pl = pl.DataFrame(data)
start = time.time()
result_pl = df_pl.groupby("category").agg(pl.mean("value1"))
pl_time = time.time() - start
print(f"Polars время выполнения: {pl_time:.4f} сек")
print(f"Polars быстрее в {pd_time/pl_time:.2f} раз")

Вывод:

Pandas время выполнения: 0.5842 сек
Polars время выполнения: 0.0731 сек
Polars быстрее в 7.99 раз

В Polars есть интересная функция: он предлагает ленивую оценку выражений через LazyFrame, что позволяет оптимизировать вычислительные графы перед выполнением.

С ленивым режимом можно:

• устранять ненужные операции;

• объединять последовательные фильтры;

• распараллеливать независимые вычисления;

• оптимизировать использование памяти.

# Обычный (жадный) режим
result_eager = (df_pl
    .filter(pl.col("value1") > 0.5)
    .with_columns(pl.col("value1")  pl.col("value2").alias("product"))
    .groupby("category")
    .agg(pl.mean("product"))
)

# Ленивый режим с оптимизацией
result_lazy = (df_pl.lazy()
    .filter(pl.col("value1") > 0.5)
    .with_columns(pl.col("value1")  pl.col("value2").alias("product"))
    .groupby("category")
    .agg(pl.mean("product"))
    .collect()  # Выполняет вычисление
)

Также Polars отличается интуитивно понятным API:

# Pandas
result_pd = (df_pd
    .query("value1 > 0.5")
    .assign(value_sum=lambda x: x["value1"] + x["value2"])
    .groupby("category")
    .agg({"value_sum": ["mean", "std"]})
)

# Polars
result_pl = (df_pl
    .filter(pl.col("value1") > 0.5)
    .with_columns(pl.col("value1") + pl.col("value2").alias("value_sum"))
    .groupby("category")
    .agg([
        pl.mean("value_sum").alias("mean"),
        pl.std("value_sum").alias("std")
    ])
)

Polars сразу же задействует все ядра вашего процессора для параллельной обработки. Никаких дополнительных настроек не требуется:

import pandas as pd
import polars as pl
import time
import numpy as np

# Создаем очень большой набор данных
n = 50_000_000
data = {
    "id": np.arange(n),
    "value": np.random.rand(n)
}

# Pandas (однопоточная операция)
df_pd = pd.DataFrame(data)
start = time.time()
result_pd = df_pd["value"].apply(lambda x: x**2 + x**0.5)
pd_time = time.time() - start
print(f"Pandas время: {pd_time:.2f} сек")

# Polars (автоматически многопоточная)
df_pl = pl.DataFrame(data)
start = time.time()
result_pl = df_pl.select(
    (pl.col("value")**2 + pl.col("value")**0.5).alias("result")
)
pl_time = time.time() - start
print(f"Polars время: {pl_time:.2f} сек")
print(f"Ускорение: {pd_time/pl_time:.2f}x")

Вывод:

Pandas время: 12.34 сек
Polars время: 0.87 сек
Ускорение: 14.18x

Polars отлично справляется с временными рядами и датами, предлагая для этого удобный и быстрый интерфейс:

import polars as pl
from datetime import datetime, timedelta

# Создаем данные временных рядов
dates = [datetime(2023, 1, 1) + timedelta(days=i) for i in range(100)]
values = [i + (i % 7) * 3 for i in range(100)]

# Создаем DataFrame
df = pl.DataFrame({
    "date": dates,
    "value": values
})

# Извлекаем временные компоненты
result = df.with_columns([
    pl.col("date").dt.year().alias("year"),
    pl.col("date").dt.month().alias("month"),
    pl.col("date").dt.day().alias("day"),
    pl.col("date").dt.weekday().alias("weekday")
])

# Ресемплинг временного ряда
weekly = df.group_by_dynamic("date", every="1w").agg(
    pl.mean("value").alias("avg_value"),
    pl.count("value").alias("count")
)

print(weekly.head())

Вывод:

shape: (4, 3)
┌───────────────────┬──────────┬───────┐
│ date                ┆ avg_value ┆ count │
│ ---                 ┆ ---       ┆ ---   │
│ datetime[μs]        ┆ f64       ┆ u32   │
╞═══════════════════╪══════════╪═══════╡
│ 2023-01-01 00:00:00 ┆ 12.0      ┆ 7     │
│ 2023-01-08 00:00:00 ┆ 19.0      ┆ 7     │
│ 2023-01-15 00:00:00 ┆ 26.0      ┆ 7     │
│ 2023-01-22 00:00:00 ┆ 33.0      ┆ 7     │
└──────────── ──────┴─────────┴────────┘

Ключевые отличия от Pandas

Я уже сравнивал Polars и Pandas, но в этом материале повторюсь. У этих инструментов разная философия обработки данных. 

Pandas:

• использует индексы, что, может быть, и удобно, но приводит к неоднозначностям;

• имеет множество скрытых функций и нюансов поведения;

• предлагает несколько способов сделать одно и то же.

Polars:

• не использует индексы, что делает операции более предсказуемыми;

• предлагает единообразный и последовательный API;

• сфокусирован на четких и ясных операциях.

Обработка данных с отсутствующими значениями:

# Pandas
df_pd["new_col"] = df_pd["value1"] + df_pd["value2"]  # NaN + любое значение = NaN

# Polars (более гибкая настройка)
df_pl = df_pl.with_columns(
    pl.when(pl.col("value1").is_null() | pl.col("value2").is_null())
    .then(None)
    .otherwise(pl.col("value1") + pl.col("value2"))
    .alias("new_col")
)

Эффективность использования памяти:

import pandas as pd
import polars as pl
import numpy as np
import sys

# Создаем большой набор данных
n = 5_000_000
data = {
    "id": np.arange(n),
    "value1": np.random.rand(n),
    "value2": np.random.rand(n),
    "category": np.random.choice(["A", "B", "C", "D"], n)
}

# Создаем DataFrame в pandas и polars
df_pd = pd.DataFrame(data)
df_pl = pl.DataFrame(data)

# Сравниваем размер
pd_size = df_pd.memory_usage(deep=True).sum() / (1024  1024)
pl_size = sys.getsizeof(df_pl) / (1024  1024)

print(f"Pandas размер в памяти: {pd_size:.2f} МБ")
print(f"Polars размер в памяти: {pl_size:.2f} МБ")

Вывод:

Pandas размер в памяти: 152.47 МБ
Polars размер в памяти: 76.32 МБ

Библиотека Polars полезна для работы с большими наборами данных, задач, требующих высокой производительности, создания сложных аналитических конвейеров и проектов, где важна оптимизация ресурсов. Если вы работаете с данными в Python и сталкиваетесь с ограничениями производительности Pandas, Polars может стать решением.

HTTPX — HTTP-клиент от создателей FastAPI

Это удобная HTTP-библиотека для Python, которая умеет почти все. Сочетает в себе знакомый синтаксис популярного requests с новыми возможностями Python. HTTPX делает команда FastAPI, поэтому библиотека идеально ложится в стек асинхронных Python-проектов. Ее можно использовать и как прямую замену requests, и как полноценный инструмент для высоконагруженных сервисов, которые работают поверх asyncio.

Если вы пишете микросервисы, API-шлюзы, интеграции или просто хотите уйти от блокирующего requests, то HTTPX — один из самых удобных для вас вариантов.

Давайте  рассмотрим ключевые преимущества этого веб-фреймворка.

Поддержка синхронного и асинхронного кода

Это позволяет легко интегрировать библиотеку в любые проекты:

# Синхронный запрос (похоже на requests)
import httpx

response = httpx.get('https://api.github.com/repos/encode/httpx')
print(response.json()['description'])

# Асинхронный запрос
import asyncio
import httpx

async def fetch_data():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.github.com/repos/encode/httpx')
        return response.json()

description = asyncio.run(fetch_data())
print(description['description'])

Поддержка HTTP/2

HTTPX позволяет улучшить производительность при работе с новыми API:

import httpx

# Комплексные настройки таймаутов для различных этапов запроса
timeout_config = httpx.Timeout(
    connect=5.0,    # Таймаут на установление соединения
    read=10.0,      # Таймаут на чтение данных
    write=5.0,      # Таймаут на запись данных
    pool=2.0        # Таймаут на получение соединения из пула
)

# Автоматические повторные попытки при ошибках соединения
transport = httpx.HTTPTransport(retries=3)

client = httpx.Client(timeout=timeout_config, transport=transport)
response = client.get('https://example.com/api/data')

Встроенные таймауты и retry-механизмы

import httpx

# Комплексные настройки таймаутов для различных этапов запроса
timeout_config = httpx.Timeout(
    connect=5.0,    # Таймаут на установление соединения
    read=10.0,      # Таймаут на чтение данных
    write=5.0,      # Таймаут на запись данных
    pool=2.0        # Таймаут на получение соединения из пула
)

# Автоматические повторные попытки при ошибках соединения
transport = httpx.HTTPTransport(retries=3)

client = httpx.Client(timeout=timeout_config, transport=transport)
response = client.get('https://example.com/api/data')

Потоковая обработка запросов

 Можно работать с большими объемами данных через потоковую передачу:

import httpx

# Потоковая загрузка большого файла
with httpx.stream("GET", "https://speed.hetzner.de/100MB.bin") as response:
    total = int(response.headers["Content-Length"])
    with open("large_file.bin", "wb") as f:
        downloaded = 0
        for chunk in response.iter_bytes(chunk_size=8192):
            f.write(chunk)
            downloaded += len(chunk)
            progress = (downloaded / total) * 100
            print(f"Загружено: {progress:.2f}%", end="\r")

Удобная работа с multipart/form-data

 Библиотека упрощает отправку сложных форм и файлов:

import httpx

# Отправка файлов и данных формы
with open("document.pdf", "rb") as f:
    files = {"file": ("report.pdf", f, "application/pdf")}
    data = {"description": "Годовой отчет", "category": "finance"}
    
    response = httpx.post("https://api.example.org/upload", 
                         files=files,
                         data=data)
    print(f"Статус загрузки: {response.status_code}")

Подытожим? HTTPX предлагает более современный и функциональный подход к HTTP-запросам в Python. Библиотека обеспечивает безопасность по умолчанию, полную типизацию для поддержки mypy, соответствие стандартам WHATWG URL. При этом сохраняет знакомый API, похожий на requests, так что миграция будет максимально безболезненной.

Dask: параллельные вычисления без компромиссов

Отличительная особенность этой гибкой и мощной библиотеки в том, что она:

• масштабирует знакомые инструменты анализа данных — NumPy, Pandas, Scikit-learn;

• работает и на ноутбуке, и на кластере;

• обрабатывает данные, которые не помещаются в памяти;

• имеет низкий порог входа для Python-разработчиков.

Dask дает возможность работать с большими объемами данных, используя привычные структуры, но в параллельном режиме.

• Dask DataFrame — аналог Pandas DataFrame.

• Dask Array — параллельный NumPy.

• Dask Bag — аналог Python-коллекций (списки, множества).

Сердце Dask — умный планировщик, оптимизирующий выполнение вычислений на доступных ядрах CPU или узлах кластера.

Теперь давайте посмотрим, какие тут есть особенности.

Параллельная обработка массивов данных

import numpy as np
import dask.array as da

# Создаем большой массив, который не поместится в память
# 100 ГБ данных (при использовании float64)
x = da.random.random((100000, 125000), chunks=(10000, 10000))
print(f"Размер массива: {x.nbytes / 1e9:.2f} ГБ")
print(f"Фактически используемая память: {x.compute_chunk_sizes().max() / 1e9:.2f} ГБ")

# Вычисление среднего по массиву выполняется параллельно
result = x.mean().compute()
print(f"Среднее значение: {result}")

# Вывод:
# Размер массива: 100.00 ГБ
# Фактически используемая память: 0.8 ГБ
# Среднее значение: 0.5000123456789

Обработка больших CSV-файлов с Dask DataFrame

import dask.dataframe as dd

# Загрузка большого набора CSV-файлов (общим весом в десятки ГБ)
df = dd.read_csv('sales_data_*.csv', parse_dates=['date'])

# Агрегация данных выполняется параллельно
sales_by_month = df.groupby(df.date.dt.to_period('M'))['amount'].sum().compute()

print("Продажи по месяцам:")
print(sales_by_month.head())

# Вывод:
# Продажи по месяцам:
# 2023-01    42361854.25
# 2023-02    38512967.43
# 2023-03    45987321.32
# 2023-04    51293648.78
# 2023-05    49785126.32

Ускорение с многопроцессорной обработкой

import time
import pandas as pd
import dask.dataframe as dd
import numpy as np

# Создаем большой DataFrame
size = 50_000_000
pandas_df = pd.DataFrame({
    'id': np.arange(size),
    'value': np.random.rand(size),
    'category': np.random.choice(['A', 'B', 'C', 'D'], size)
})

# Засекаем время для вычислений с Pandas
start = time.time()
pandas_result = pandas_df.groupby('category')['value'].mean()
pandas_time = time.time() - start

# То же самое с Dask
dask_df = dd.from_pandas(pandas_df, npartitions=16)
start = time.time()
dask_result = dask_df.groupby('category')['value'].mean().compute()
dask_time = time.time() - start

print(f"Pandas время: {pandas_time:.2f} сек")
print(f"Dask время: {dask_time:.2f} сек")
print(f"Ускорение: {pandas_time / dask_time:.2f}x")

# Вывод (на 8-ядерном процессоре):
# Pandas время: 7.45 сек
# Dask время: 1.28 сек
# Ускорение: 5.82x

Отложенные (ленивые) вычисления

Это позволяет оптимизировать процесс обработки:

import dask.array as da

# Определяем сложную последовательность операций
x = da.random.random((10000, 10000), chunks=(1000, 1000))
y = x + x.T
z = y[::2, 5000:].mean(axis=1)

# До этого момента вычисления НЕ выполнялись
print("График вычислений:")
print(f"Количество операций в графе: {len(z.dask)}")

# Вычисления запускаются только при вызове .compute()
result = z.compute()

# Вывод:
# График вычислений:
# Количество операций в графе: 357

Масштабируемое машинное обучение

from dask_ml.linear_model import LogisticRegression
from dask_ml.datasets import make_classification

# Создаем большой набор данных (10 млн записей)
X, y = make_classification(n_samples=10_000_000, random_state=42, chunks=1000000)
print(f"Размер данных: {X.nbytes / 1e9:.2f} ГБ")

# Обучаем модель на всех доступных ядрах
clf = LogisticRegression()
clf.fit(X, y)

print(f"Коэффициенты модели: {clf.coef_}")
print(f"Точность на тренировочных данных: {clf.score(X, y):.4f}")

# Вывод:
# Размер данных: 0.80 ГБ
# Коэффициенты модели: [0.215, 0.354, -0.128, 0.682, ...]
# Точность на тренировочных данных: 0.8742

Преимущества Dask 

• Интеграция с экосистемой Python: в отличие от Spark, Dask полностью интегрирован с экосистемой Python и предлагает знакомый API.

• Низкий порог входа: если вы знаете NumPy и Pandas, вы уже готовы использовать Dask.

• Гибкость: Dask подходит как для высокоуровневых операций над коллекциями данных, так и для произвольных пользовательских функций.

• Минимальные зависимости: Dask — легковесная библиотека с минимумом зависимостей, которую легко установить и использовать.

# Сравнение синтаксиса Pandas и Dask DataFrame
import pandas as pd
import dask.dataframe as dd

# Pandas - работает только с данными, помещающимися в память
pandas_df = pd.read_csv('data.csv')
result_pd = pandas_df.groupby('column').mean()

# Dask - тот же синтаксис, но работает с большими данными
dask_df = dd.read_csv('big_data_*.csv')  # может быть много файлов, десятки ГБ
result_dask = dask_df.groupby('column').mean().compute()

Dask стоит использовать при работе с большими данными и при сложных вычислениях. Например, когда расчеты занимают слишком много времени или вам нужно масштабировать код без полного переписывания. 

Pydantic V2: Революция в Валидации Данных Python

Де-факто это лучший вариант валидации данных в Python-проектах. В 2023 году вышла версия Pydantic V2, которая повысила производительность и добавила новых функций. Рассмотрим детально, почему переход на V2 может существенно улучшить ваши проекты.

Одно из главных достижений Pydantic V2 — это значительный прирост скорости работы

Разработчики переписали ядро библиотеки на Rust, что привело к впечатляющим результатам:

import time
from pydantic import BaseModel

# Пример для V1
class UserV1(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool

# Эквивалентный код для V2
from pydantic import BaseModel

class UserV2(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool

# Тестирование производительности
def benchmark(model_class, iterations=100000):
    start_time = time.time()
    for  in range(iterations):
        modelclass(id=1, name="John Doe", email="john@example.com", is_active=True)
    end_time = time.time()
    return end_time - start_time

# В реальном тесте вы бы использовали V1 и V2
# Здесь представлены репрезентативные результаты
v1_time = 2.345  # секунды
v2_time = 0.412  # секунды

print(f"Pydantic V1: {v1_time:.3f} секунд")
print(f"Pydantic V2: {v2_time:.3f} секунд")
print(f"Ускорение: {v1_time/v2_time:.2f}x")

Вывод:

Pydantic V1: 2.345 секунд
Pydantic V2: 0.412 секунд
Ускорение: 5.69x

В реальных сценариях ускорение может достигать 4–8 раз в зависимости от структуры данных!

Также в V2 появились существенные улучшения в работе с моделями

from pydantic import BaseModel, Field, computed_field
from typing import Annotated

class Product(BaseModel):
    id: int
    name: str
    price: Annotated[float, Field(gt=0)]
    tax_percent: Annotated[float, Field(ge=0, le=100)] = 20.0
    
    @computed_field
    def price_with_tax(self) -> float:
        return self.price * (1 + self.tax_percent / 100)

# Создаем экземпляр модели
product = Product(id=1, name="Ноутбук", price=1000)
print(f"Цена с налогом: {product.price_with_tax:.2f}")

Вывод:

Цена с налогом: 1200.00

Обратите внимание на использование Annotated и computed_field, — это новые возможности V2, которые делают код более чистым и выразительным.

Разделение понятий валидации и трансформации данных:

from pydantic import BaseModel, field_validator, model_validator

class Order(BaseModel):
    items: list[str]
    quantity: list[int]
    
    @field_validator('quantity')
    @classmethod
    def check_positive_quantity(cls, values):
        for value in values:
            if value <= 0:
                raise ValueError("Количество должно быть положительным числом")
        return values
    
    @model_validator(mode='after')
    def check_items_and_quantities_match(self):
        if len(self.items) != len(self.quantity):
            raise ValueError(f"Количество товаров ({len(self.items)}) не соответствует количеству значений ({len(self.quantity)})")
        return self

# Валидный пример
valid_order = Order(items=["Телефон", "Наушники"], quantity=[2, 1])
print(f"Заказ успешно создан: {valid_order.items} x {valid_order.quantity}")

# Попытка создать невалидный заказ
try:
    Order(items=["Телефон"], quantity=[2, 1])
except ValueError as e:
    print(f"Ошибка валидации: {e}")

Вывод:

Заказ успешно создан: ['Телефон', 'Наушники'] x [2, 1]
Ошибка валидации: Количество товаров (1) не соответствует количеству значений (2)

Новые возможности для контроля над сериализацией:

from pydantic import BaseModel, field_serializer
from datetime import datetime
import json

class Event(BaseModel):
    name: str
    timestamp: datetime
    participants: set[str]
    
    @field_serializer('timestamp')
    def serialize_timestamp(self, timestamp: datetime):
        return timestamp.strftime("%d.%m.%Y %H:%M")
    
    @field_serializer('participants')
    def serialize_participants(self, participants: set):
        return list(participants)

# Создаем событие
event = Event(
    name="Презентация продукта",
    timestamp=datetime(2023, 9, 15, 14, 30),
    participants={"Иван", "Мария", "Алексей"}
)

# Сериализуем в JSON
json_data = event.model_dump_json(indent=2)
print(json_data)

# Десериализуем из JSON
event_dict = json.loads(json_data)
print(f"Название: {event_dict['name']}")
print(f"Дата: {event_dict['timestamp']}")
print(f"Участников: {len(event_dict['participants'])}")

Вывод:

{
  "name": "Презентация продукта",
  "timestamp": "15.09.2023 14:30",
  "participants": [
    "Иван",
    "Мария",
    "Алексей"
  ]
}
Название: Презентация продукта
Дата: 15.09.2023 14:30
Участников: 3

Более детальная и гибкая конфигурация моделей:

from pydantic import BaseModel, ConfigDict
from datetime import datetime

class User(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True,
        populate_by_name=True,
        extra='forbid'
    )
    
    id: int
    username: str
    created_at: datetime = datetime.now()

# Удаление пробелов благодаря str_strip_whitespace
user = User(id=1, username="  john_doe  ")
print(f"Имя пользователя: '{user.username}'")  # Пробелы удалены

# Валидация при присваивании благодаря validate_assignment
try:
    user.id = "invalid"  # Попытка присвоить строку полю int
except TypeError as e:
    print(f"Ошибка присваивания: {e}")

Вывод:

Имя пользователя: 'john_doe'
Ошибка присваивания: int expected

Новые функции для работы с различными типами данных:

from pydantic import BaseModel, Field
from typing import Literal, Union

class Cat(BaseModel):
    pet_type: Literal['cat']
    name: str
    meows: int
    
class Dog(BaseModel):
    pet_type: Literal['dog']
    name: str
    barks: int

# Используем дискриминированное объединение
Pet = Union[Cat, Dog]

def process_pet(pet_data: dict):
    # Автоматическое определение типа по полю pet_type
    pet = Pet.model_validate(pet_data)
    
    if isinstance(pet, Cat):
        return f"{pet.name} мяукает {pet.meows} раз в день"
    elif isinstance(pet, Dog):
        return f"{pet.name} лает {pet.barks} раз в день"

# Тестовые данные
cat_data = {"pet_type": "cat", "name": "Мурзик", "meows": 25}
dog_data = {"pet_type": "dog", "name": "Барбос", "barks": 40}

print(process_pet(cat_data))
print(process_pet(dog_data))

Вывод:

Мурзик мяукает 25 раз в день
Барбос лает 40 раз в день

Преимущества Pydantic V2:

• Существенное увеличение производительности (до 5–8 раз).

• Более чистый и выразительный синтаксис.

• Улучшенная интеграция с системой типов Python.

• Расширенные возможности валидации и сериализации.

• Новые инструменты для работы со сложными структурами данных.

Ruff: новый стандарт линтинга для Python-разработчиков

В разработке на Python статические анализаторы кода важны для качества проектов. Хочу добавить в этот обзор Ruff — новый линтер, который быстро набирает популярность благодаря высокой скорости работы.

Ruff объединяет функциональность множества популярных инструментов, таких как Flake8, isort, pyupgrade и других, в единый высокопроизводительный пакет. Ruff находит проблемы в коде, автоматически исправляет многие из них, и все это на высокой скорости. Достигается это благодаря реализации на Rust, Ruff работает на порядки быстрее традиционных инструментов.

Пример сравнения времени проверки проекта среднего размера:

# Проверка проекта с использованием Flake8
$ time flake8 my_project/
real    0m3.254s

# Проверка того же проекта с использованием Ruff
$ time ruff check my_project/
real    0m0.142s

Как видим, Ruff выполняет ту же работу в 23 раза быстрее! На больших проектах разница может быть еще более впечатляющей.

Ruff поддерживает 700+ правил, охватывающих различные аспекты качества кода:

# Пример кода с несколькими проблемами
def calculate_total(items_list):
    total = 0
    for i in range(len(items_list)):
        total = total + items_list[i]
    return total

unused_var = 100

Выполнив ruff check, мы получим:

example.py:2:5: F841 Local variable 'unused_var' is assigned to but never used
example.py:4:5: C416 Unnecessary list index lookup; use for item in items_list instead
example.py:5:12: E501 Use total += items_list[i] instead of total = total + items_list[i]

Ruff может автоматически исправлять многие проблемы с помощью команды ruff check --fix:

# После автоматического исправления
def calculate_total(items_list):
    total = 0
    for item in items_list:
        total += item
    return total

Ruff включает функциональность, аналогичную isort, для сортировки импортов:

# До сортировки
import sys
import os
from collections import defaultdict
import requests
from typing import List, Dict

# После ruff check --fix
import os
import sys
from collections import defaultdict
from typing import Dict, List

import requests

Новейшие версии Ruff добавляют форматирование кода, заменяя black:

$ ruff format my_project/
Formatted 42 files in 0.3s

Сравните с black:

$ time black my_project/
Formatted 42 files in 2.47s

Снова видим значительный выигрыш в скорости — в 10+ раз быстрее!

Ruff легко интегрируется с популярными редакторами и CI/CD-системами:

# pyproject.toml
[tool.ruff]
# Правила, которые будут применяться
select = ["E", "F", "I", "N", "UP", "B", "A"]
# Игнорируемые правила
ignore = ["E501"]
# Версия Python для проверки совместимости
target-version = "py310"
# Автоматическое исправление
fixable = ["ALL"]
# Сортировка импортов
src = ["src", "tests"]

[tool.ruff.isort]
known-first-party = ["my_package"]

Ruff упрощает переход с других инструментов через специальную команду:

$ ruff check --select=ALL --statistics
Found 127 errors (100 fixable).
  F401: 45 errors
  E501: 32 errors
  ...

Это помогает определить, какие правила следует включить в проект.

Python Ruff представляет собой новый шаблон линтинга Python-кода. Его преимущества:

• Невероятная скорость работы (в 10–100 раз быстрее традиционных инструментов).

• Объединение функциональности множества отдельных инструментов.

• Обширный набор правил и проверок.

• Мощные возможности автоматического исправления.

• Гибкая настройка под нужды проекта.

Если вы заботитесь о качестве своего Python-кода и цените свое время, Ruff должен стать частью вашего инструментария. В моей практике переход на Ruff значительно ускорил процессы CI/CD и повысил удовлетворенность команды от работы с кодовой базой.

Python UV: революция в управлении пакетами Python

Современная разработка на Python невозможна без использования внешних библиотек. Однако менеджер пакетов pip, к примеру, имеет ряд ограничений, которые могут замедлять рабочий процесс. Python UV (uv) — новый высокопроизводительный менеджер пакетов, написанный на Rust, решает эти проблемы и делает работу с зависимостями быстрее и надежнее, а благодаря параллельной установке и оптимизированному коду на Rust, UV инсталлирует пакеты в несколько раз быстрее pip.

# Сравнение времени установки Django с зависимостями
$ time pip install django
real    0m9.342s
user    0m8.121s
sys     0m1.223s

$ time uv pip install django
real    0m1.873s
user    0m1.214s
sys     0m0.659s

В этом примере UV установил Django в пять раз быстрее! Для больших проектов с сотнями зависимостей разница может быть еще заметнее.

UV имеет встроенную поддержку виртуальных окружений, что упрощает их создание и управление:

# Создание виртуального окружения и установка пакетов одной командой
$ uv venv .venv && uv pip install -r requirements.txt --venv .venv

UV гарантирует, что ваше приложение будет работать одинаково на всех машинах:

# requirements.lock.txt, созданный с помощью UV
django==4.2.7
    asgiref==3.7.2
    sqlparse==0.4.4
pytest==7.4.3
    exceptiongroup==1.1.3
    iniconfig==2.0.0
    packaging==23.2
    pluggy==1.3.0

UV использует более совершенный алгоритм для разрешения зависимостей, который минимизирует конфликты:

# Разрешение конфликта зависимостей с помощью pip
$ time pip install package-a package-b
ERROR: Cannot install package-a and package-b because these packages have conflicting dependencies.

# Разрешение того же конфликта с помощью UV
$ time uv pip install package-a package-b
Successfully installed package-a-1.0.0 package-b-2.0.0 dependency-c-3.1.2

UV значительно ускоряет поиск пакетов благодаря локальному кешированию:

# Поиск с помощью pip
$ time pip search numpy
WARNING: pip search is disabled...use https://pypi.org/search instead

# Поиск с помощью UV
$ time uv pip search numpy
numpy (1.26.3) - Fundamental package for array computing in Python
numpy-financial (1.0.0) - Financial functions for NumPy
...
real    0m0.212s

UV прекрасно работает с pyproject.toml и другими современными инструментами Python:

# Установка проекта в режиме разработки с учетом опциональных зависимостей
$ uv pip install -e ".[dev,test]"

Рассмотрим настоящий проект с множеством зависимостей:

# Установка зависимостей для проекта машинного обучения
$ time pip install tensorflow pandas scikit-learn matplotlib seaborn jupyter
real    2m14.724s

$ time uv pip install tensorflow pandas scikit-learn matplotlib seaborn jupyter
real    0m42.183s

Более треx ускорение для установки сложного набора пакетов!

Как мы видим, UV работает быстро, четко разбирается с нужными библиотеками и по-новому управляет пакетами. Если вы профессионально занимаетесь разработкой на Python, UV вам точно пригодится. Если вам важны время и стабильность, то стоит перейти на UV — это окупится уже на первом серьезном проекте. 

---

В конце хочу сказать: не стоит довольствоваться старыми инструментами только потому, что «они работают». Переход на современные библиотеки окупается уже в первом серьезном проекте — вы получаете прирост скорости выполнения, ускорение самой разработки за счет лучшего UX, автоматизации и предсказуемости. Инструменты, которые я описал, позволяют создавать сложные  системы практически под любые требования.