Профилирование, поиск bottleneck, benchmark, оптимизация
Производительность критична для высокой нагрузки. В этой теме вы научитесь профилированию, поиску bottleneck, benchmark и оптимизации кода.
Не оптимизируйте без измерений!
import cProfile
import pstats
from pstats import SortKey
def main():
# Ваш код
for i in range(1000):
result = i ** 2
return result
# Запуск профилировщика
profiler = cProfile.Profile()
profiler.enable()
main()
profiler.disable()
# Статистика
stats = pstats.Stats(profiler)
stats.sort_stats(SortKey.TIME)
stats.print_stats(10) # Топ 10 функций 1007 function calls in 0.002 seconds
Ordered by: internal time
ncalls tottime percall cumtime percall filename:lineno(function)
1000 0.001 0.000 0.001 0.000 main.py:5(<module>)
1 0.000 0.000 0.002 0.002 main.py:4(main)
1 0.000 0.000 0.000 0.000 {method 'disable' of '_lsprof.Profiler' objects}
| Столбец | Описание |
|---|---|
ncalls | Количество вызовов |
tottime | Время в функции (без вызовов других) |
percall | Среднее время на вызов |
cumtime | Кумулятивное время (с вызовами других) |
pip install line_profilerfrom line_profiler import LineProfiler
def slow_function():
total = 0
for i in range(1000000): # Медленно!
total += i
return total
def fast_function():
return sum(range(1000000)) # Быстро!
# Профилирование
profiler = LineProfiler()
profiler.add_function(slow_function)
profiler.add_function(fast_function)
wrapper = profiler(slow_function)
wrapper()
profiler.print_stats()Timer unit: 1e-06 s
Total time: 0.05 s
File: main.py
Function: slow_function at line 3
Line # Hits Time Per Hit % Time Line Contents
==============================================================
3 def slow_function():
4 1 1.0 1.0 0.0 total = 0
5 1000001 49000.0 0.0 98.0 for i in range(1000000):
6 1000000 900.0 0.0 1.8 total += i
7 1 1.0 1.0 0.0 return total
# 1000 запросов, 10 параллельных
ab -n 1000 -c 10 http://localhost:8000/items
# Вывод
Requests per second: 5000.00 [#/sec]
Time per request: 2.000 [ms]# 12 потоков, 400 подключений, 30 секунд
wrk -t12 -c400 -d30s http://localhost:8000/itemspip install locustlocustfile.py:
from locust import HttpUser, task, between
class APIUser(HttpUser):
wait_time = between(1, 3)
@task(3)
def get_items(self):
self.client.get("/items")
@task(1)
def get_user(self):
self.client.get("/users/1")
@task(2)
def create_item(self):
self.client.post("/items", json={
"name": "Test Item",
"price": 99.99
})Запуск:
locust -f locustfile.py --host=http://localhost:8000
# Откройте http://localhost:8089| Место | Симптомы | Решение |
|---|---|---|
| База данных | Медленные запросы, >100ms | Индексы, кэширование, оптимизация запросов |
| Внешние API | Таймауты, >500ms | Кэширование, timeout, retry |
| CPU | 100% загрузка | Оптимизация алгоритмов, multiprocessing |
| Память | Утечки, GC | Профилирование памяти, оптимизация структур |
| I/O | Блокировки | Async, thread pool |
from sqlalchemy import event
import time
@event.listens_for(engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
conn.info.setdefault('query_start_time', []).append(time.time())
@event.listens_for(engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
total = time.time() - conn.info['query_start_time'].pop(-1)
print(f"Query time: {total:.3f}s")
print(f"Query: {statement}")from sqlalchemy.orm import Session, joinedload, selectinload
from sqlalchemy import select
# Плохо: N+1 запрос
users = session.execute(select(User)).scalars().all()
for user in users:
print(user.posts) # Запрос к БД для каждого!
# Хорошо: 1 запрос с join
users = session.execute(
select(User).options(joinedload(User.posts))
).scalars().all()
for user in users:
print(user.posts) # Без дополнительных запросов# Медленно
result = []
for i in range(1000):
result.append(i * 2)
# Быстро
result = list(map(lambda x: x * 2, range(1000)))
# Ещё быстрее
result = [i * 2 for i in range(1000)]# Список (память O(n))
def get_squares(n):
return [i ** 2 for i in range(n)]
# Генератор (память O(1))
def get_squares(n):
for i in range(n):
yield i ** 2
# Использование
for square in get_squares(1000000):
pass # Не хранит все в памятиfrom functools import lru_cache
@lru_cache(maxsize=128)
def expensive_computation(n):
time.sleep(1) # Имитация
return n ** 2
# Первый вызов: 1 секунда
expensive_computation(10)
# Второй вызов: мгновенно (из кэша)
expensive_computation(10)# Медленно: поиск в списке O(n)
if item in my_list:
...
# Быстро: поиск в множестве O(1)
my_set = set(my_list)
if item in my_set:
...# Медленно: O(n²)
result = ""
for i in range(1000):
result += str(i)
# Быстро: O(n)
result = "".join(str(i) for i in range(1000))import httpx
import asyncio
# Последовательно (3 секунды)
async def sequential():
async with httpx.AsyncClient() as client:
r1 = await client.get('https://api1.com')
r2 = await client.get('https://api2.com')
r3 = await client.get('https://api3.com')
return [r1, r2, r3]
# Параллельно (1 секунда)
async def parallel():
async with httpx.AsyncClient() as client:
r1, r2, r3 = await asyncio.gather(
client.get('https://api1.com'),
client.get('https://api2.com'),
client.get('https://api3.com')
)
return [r1, r2, r3]import httpx
# Один клиент на всё приложение (connection pool)
client = httpx.AsyncClient()
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app):
yield
await client.aclose()
app = FastAPI(lifespan=lifespan)
@app.get('/external')
async def get_external():
return await client.get('https://api.example.com')from sqlalchemy import Index
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String, index=True)
username: Mapped[str] = mapped_column(String, index=True, unique=True)
__table_args__ = (
Index('ix_users_email_username', 'email', 'username'),
)# Плохо: загрузка всех записей
users = session.execute(select(User)).scalars().all()
return users[0:10]
# Хорошо: пагинация на уровне БД
users = session.execute(select(User).offset(0).limit(10)).scalars().all()
return usersfrom sqlalchemy.orm import selectinload
# N+1 запрос
users = session.execute(select(User)).scalars().all()
for user in users:
posts = session.execute(select(Post).where(Post.user_id == user.id)).scalars().all()
# 2 запроса
users = session.execute(
select(User).options(selectinload(User.posts))
).scalars().all()
for user in users:
posts = user.posts # Без запросов# Pydantic v1
class User(BaseModel):
...
# Pydantic v2 (быстрее в 5-50 раз)
class User(BaseModel):
model_config = ConfigDict(...)user = User(...)
# Медленнее
data = dict(user)
# Быстрее
data = user.model_dump()pip install prometheus-fastapi-instrumentatorfrom fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
app = FastAPI()
Instrumentator().instrument(app).expose(app)
# Метрики доступны на /metricsfrom fastapi import FastAPI, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from functools import lru_cache
from contextlib import asynccontextmanager
import httpx
import asyncio
# Кэширование конфигурации
@lru_cache(maxsize=1)
def get_config():
return {
'external_api_url': 'https://api.example.com',
'cache_ttl': 300
}
# Connection pool для внешних запросов
http_client = httpx.AsyncClient()
@asynccontextmanager
async def lifespan(app):
yield
await http_client.aclose()
app = FastAPI(lifespan=lifespan)
@app.get('/users/{user_id}/dashboard')
async def get_dashboard(
user_id: int,
db: AsyncSession = Depends(get_db),
include_stats: bool = Query(False),
config: dict = Depends(get_config)
):
"""
Оптимизированный endpoint dashboard.
"""
# 1. Загрузка пользователя с связями (1 запрос вместо N+1)
user_result = await db.execute(
select(User)
.options(selectinload(User.posts), selectinload(User.profile))
.where(User.id == user_id)
)
user = user_result.scalar_one_or_none()
if not user:
raise HTTPException(404, "User not found")
# 2. Параллельные запросы к внешнему API
external_data = None
if include_stats:
try:
# Таймаут 3 секунды
external_data = await asyncio.wait_for(
http_client.get(f"{config['external_api_url']}/users/{user_id}/stats"),
timeout=3.0
)
external_data = external_data.json()
except asyncio.TimeoutError:
external_data = {'error': 'timeout'}
except Exception as e:
external_data = {'error': str(e)}
# 3. Пагинация для постов
posts_result = await db.execute(
select(Post)
.where(Post.user_id == user_id)
.order_by(Post.created_at.desc())
.offset(0)
.limit(10)
)
posts = posts_result.scalars().all()
return {
'user': user,
'posts': posts,
'external_stats': external_data
}Лаборатория
«N+1, пагинация, batch loading»
доступна в публичном GitLab. Она проверяет устранение N+1 через
selectinload/joinedload, пагинацию на уровне БД через offset/limit,
batch loading для связанных объектов и измерение количества запросов.
Склонируйте репозиторий и запустите тесты из его корня:
git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest performance/testsДалее: Безопасность