Когда в 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, автоматизации и предсказуемости. Инструменты, которые я описал, позволяют создавать сложные системы практически под любые требования.
Источник: habr.com
