Алгоритмическая сложность, I/O операции, N+1 проблемы, оптимизации
Ревьюер не измеряет скорость — он замечает код, который перестанет работать при росте данных. Это разные задачи, и вторая решается чтением.
Вы научитесь находить в диффе то, что деградирует нелинейно: запросы в цикле, вложенные обходы, чтение всего в память, отсутствие таймаутов. И — не менее важно — перестанете писать замечания о производительности там, где выигрыш измеряется микросекундами на сто вызовов в день.
Кэширование, очереди, backpressure и rate limiting — в теме Масштабируемость курса Pro.
Оптимизация без измерения — источник худшего кода в проекте. Поэтому у замечания о производительности должно быть обоснование в терминах роста, а не в терминах «так быстрее».
Стоит комментария:
Не стоит комментария: замена генератора списка циклом, += вместо join, микрооптимизация в коде, выполняющемся раз в сутки. Такие правки усложняют чтение, а выигрыш неизмерим.
Простой фильтр перед тем, как писать комментарий: сколько раз это выполнится и на каком объёме данных. Если ответ «десять раз на десяти элементах» — промолчите.
Проверьте себя. Просмотрите свои последние замечания о производительности. У скольких была оценка объёма?
Частая ошибка. Требовать оптимизацию по привычке к паттерну, а не по оценке нагрузки. Это тратит время автора и портит читаемость.
Главный вопрос — как время растёт с ростом входа, а не сколько операций внутри.
# ❌ O(n·m): для каждого заказа обходим весь список пользователей
def enrich(orders, users):
for order in orders:
for user in users:
if user.id == order.user_id:
order.user_name = user.name
# ✅ O(n+m): один проход на индекс, один на присваивание
def enrich(orders, users):
by_id = {user.id: user for user in users}
for order in orders:
user = by_id.get(order.user_id)
if user is not None:
order.user_name = user.nameНа сотне элементов разницы нет. На десяти тысячах заказов и десяти тысячах пользователей первый вариант делает сто миллионов сравнений — это минуты.
Самый частый и самый дешёвый в исправлении случай — проверка вхождения:
if item in big_list: # ❌ O(n) на каждую проверку
if item in big_set: # ✅ O(1)Если big_list строится один раз, а проверок много, превращение его в множество меняет сложность всего блока.
Проверьте себя. Найдите в проекте вложенный цикл по двум коллекциям и оцените, чем ограничены их размеры.
Частая ошибка. Смотреть на количество строк вместо количества итераций. Три строки в цикле по цикле хуже тридцати строк подряд.
Один запрос за списком, затем по запросу на каждый элемент.
# ❌ 1 + N запросов
orders = Order.objects.all()
for order in orders:
print(order.user.name) # каждый .user — отдельный SELECT
# ✅ 2 запроса
orders = Order.objects.select_related("user")Коварство N+1 в том, что в коде он невидим: order.user выглядит как обращение к полю. Признак в диффе — обращение к связанному объекту или вызов функции внутри цикла по результату запроса.
То же самое в SQLAlchemy — joinedload или selectinload; в чистом SQL — один запрос с IN вместо цикла:
# ❌
for order_id in order_ids:
items[order_id] = db.execute("SELECT * FROM items WHERE order_id = %s", (order_id,))
# ✅
rows = db.execute("SELECT * FROM items WHERE order_id = ANY(%s)", (order_ids,))
items = groupby_order_id(rows)И то же самое с HTTP-запросами — здесь цена ещё выше, потому что к задержке сети добавляется установка соединения:
# ❌ сто последовательных запросов
for user_id in user_ids:
results.append(requests.get(f"{API}/users/{user_id}").json())
# ✅ параллельно, с ограничением одновременности
async with aiohttp.ClientSession() as session:
sem = asyncio.Semaphore(10)
async def fetch(uid):
async with sem, session.get(f"{API}/users/{uid}") as r:
return await r.json()
results = await asyncio.gather(*(fetch(uid) for uid in user_ids))Обратите внимание на семафор: параллельность без ограничения — это способ положить чужой сервис и получить блокировку по адресу.
Проверьте себя. Включите логирование SQL в тестах и посмотрите, сколько запросов делает ваш самый нагруженный эндпоинт.
Частая ошибка. Чинить N+1 добавлением кэша. Кэш скроет проблему в тёплом состоянии и оставит её при холодном старте — то есть ровно в момент выкатки.
Код, который читает всё, работает ровно до того дня, когда данных станет много.
# ❌ весь файл в память
content = open("access.log").read()
for line in content.split("\n"):
...
# ✅ построчно
with open("access.log") as f:
for line in f:
...# ❌ вся таблица в память
users = User.objects.all()
for user in users:
...
# ✅ порциями
for user in User.objects.iterator(chunk_size=1000):
...Тот же принцип — в возвращаемых значениях: генератор вместо списка, если результат потребляется по одному.
def get_squares(n):
return [x ** 2 for x in range(n)] # ❌ n значений в памяти
def get_squares(n):
for x in range(n):
yield x ** 2 # ✅ O(1) памятиОтдельно смотрите на эндпоинты без пагинации: GET /orders без limit — это обещание однажды отдать всю таблицу.
Проверьте себя. Найдите в API эндпоинт, возвращающий список без ограничения размера, и прикиньте, каким он станет через год.
Частая ошибка. Тестировать на сотне записей. Проблема этого класса не воспроизводится на тестовых данных — она видна только чтением.
Отсутствующий таймаут — не проблема производительности одного запроса, а способ исчерпать пул соединений и уронить сервис целиком.
requests.get(url) # ❌ ждать может бесконечно
requests.get(url, timeout=(3, 10)) # ✅ на соединение и на чтениеПроверять надо каждый выход во внешний мир: HTTP-клиент, драйвер БД, кэш, очередь. И спрашивать, что происходит по истечении таймаута: повтор с задержкой, деградация или ошибка пользователю — любой ответ подходит, кроме отсутствующего.
Проверьте себя. Есть ли в проекте место, где HTTP-клиент создаётся без таймаута по умолчанию?
Частая ошибка. Ставить таймаут только на чтение и забывать про установку соединения — а зависает чаще именно она.
Новый эндпоинт для личного кабинета. На тестовых данных отвечает за 40 мс.
+@app.get("/seller/{seller_id}/stats")
+def seller_stats(seller_id: int):
+ products = Product.objects.filter(seller_id=seller_id)
+
+ total_revenue = 0
+ best = None
+ for product in products:
+ orders = Order.objects.filter(product_id=product.id)
+ revenue = sum(o.price * o.qty for o in orders)
+ total_revenue += revenue
+ if best is None or revenue > best[1]:
+ best = (product.name, revenue)
+
+ reviews = Review.objects.all()
+ my_reviews = [r for r in reviews if r.product_id in [p.id for p in products]]
+ rating = sum(r.score for r in my_reviews) / len(my_reviews)
+
+ return {"revenue": total_revenue, "best": best, "rating": rating}Что видит ревьюер. Три отдельные проблемы, каждая из которых незаметна на десяти товарах.
Первая: Order.objects.filter внутри цикла по товарам — N+1. У продавца с тысячей позиций это тысяча запросов.
Вторая: Review.objects.all() загружает все отзывы всего маркетплейса в память, чтобы отфильтровать их в Python. Плюс [p.id for p in products] пересобирается на каждой итерации списочного включения — то есть список товаров строится столько раз, сколько всего отзывов в системе. Это произведение двух больших чисел.
Третья: len(my_reviews) равен нулю у продавца без отзывов — деление на ноль. Это уже не производительность, а падение, и в комментарии это надо разделить.
Комментарии в PR:
stats.py:7· blockerReview.objects.all()тянет все отзывы платформы в память приложения — на проде это миллионы строк. Фильтрация и агрегирование должны происходить в БД:Review.objects.filter(product__seller_id=seller_id).aggregate(Avg("score")).
stats.py:15· blockerlen(my_reviews)= 0 у нового продавца →ZeroDivisionError. Страница личного кабинета упадёт у всех, кто ещё не получил отзыв.
stats.py:7· major Запрос заказов внутри цикла по товарам — N+1. Одна агрегация сGROUP BY product_idдаёт и общую выручку, и лучший товар за один запрос.
stats.py:14· major[p.id for p in products]внутри списочного включения пересчитывается на каждый отзыв. Даже если оставить фильтрацию в Python, список id надо вынести в переменную. Но лучше не оставлять.
stats.py:2· question Сколько товаров у самого крупного продавца сейчас? От этого зависит, хватит ли агрегации или нужен предпосчёт. Если это тысячи — стоит обсудить материализованное представление, но отдельным тикетом.
Чем закончилось. Три запроса вместо тысячи с лишним, ответ за 60 мс на реальных данных крупного продавца вместо 40 секунд:
@app.get("/seller/{seller_id}/stats")
def seller_stats(seller_id: int):
revenue_by_product = (
Order.objects
.filter(product__seller_id=seller_id)
.values("product__name")
.annotate(revenue=Sum(F("price") * F("qty")))
.order_by("-revenue")
)
rows = list(revenue_by_product)
total = sum(r["revenue"] for r in rows)
best = (rows[0]["product__name"], rows[0]["revenue"]) if rows else None
rating = (
Review.objects
.filter(product__seller_id=seller_id)
.aggregate(avg=Avg("score"))["avg"]
)
return {"revenue": total, "best": best, "rating": rating}Avg в БД возвращает None вместо падения — деление на ноль исчезло само вместе с переносом агрегации в базу. Так бывает часто: правильное решение по производительности заодно убирает баг.
СЛОЖНОСТЬ нет вложенных обходов по коллекциям неограниченного размера
ЗАПРОСЫ ничего не запрашивается внутри цикла — ни БД, ни HTTP, ни кэш
АГРЕГАЦИЯ считается в БД, а не в памяти приложения
ОБЪЁМ нет чтения всего файла/таблицы целиком; списки пагинируются
ПОИСК множество или словарь вместо линейного поиска по большому списку
ТАЙМАУТЫ есть на каждом внешнем вызове, включая соединение
ОЦЕНКА для каждого замечания названы объём данных и частота вызоваТренировка: Code Review Python → Производительность и Code Review React → Производительность. Индексы, планы запросов и границы транзакций — в теме Работа с БД курса Pro.
Ключевая мысль: ревьюер ищет не медленный код, а код, время работы которого растёт быстрее, чем данные.
Далее: Проверка тестов