Как можно влиять на скорость работы Python-приложения?

Оптимизация скорости Python-приложения

Это не магия, а системный подход. За 10+ лет я оптимизировал сотни приложений. Начинаю с профилирования, потом оптимизирую узкие места.

1. Профилирование — первый шаг

Не гадай, где медленно. Измеряй:

import cProfile
import pstats
import io

def slow_function():
    result = 0
    for i in range(1000000):
        result += i
    return result

# Профилирование
pr = cProfile.Profile()
pr.enable()
slow_function()
pr.disable()

s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(10)  # Топ-10 функций
print(s.getvalue())

Альтернативы:

• line_profiler — построчное профилирование
• memory_profiler — потребление памяти
• py-spy — профайлер в production

# Установка и использование
pip install line-profiler
kernprof -l -v script.py

2. Оптимизация алгоритмов — Big O

Это часто даёт 10-100x улучшение:

# ❌ O(n²) — МЕДЛЕННО
def find_duplicates_slow(items):
    result = []
    for i in range(len(items)):
        for j in range(i + 1, len(items)):
            if items[i] == items[j]:
                result.append(items[i])
    return result

# ✅ O(n) — БЫСТРО (100x быстрее для 10k элементов)
def find_duplicates_fast(items):
    seen = set()
    duplicates = set()
    for item in items:
        if item in seen:
            duplicates.add(item)
        seen.add(item)
    return list(duplicates)

3. Структуры данных имеют значение

Выбор структуры влияет на производительность:

import timeit

# Поиск в списке O(n)
data_list = list(range(10000))
time_list = timeit.timeit(
    lambda: 9999 in data_list,
    number=10000
)  # ~0.5 сек

# Поиск в set O(1)
data_set = set(range(10000))
time_set = timeit.timeit(
    lambda: 9999 in data_set,
    number=10000
)  # ~0.00001 сек (в 50,000 раз быстрее!)

print(f"List: {time_list:.4f}s, Set: {time_set:.4f}s")

Правило большого пальца:

• Частые поиски → set, dict
• Частые вставки в начало → deque
• Отсортированные данные → bisect
• Приоритет → heapq

4. Кеширование результатов

Если функция вызывается много раз с одинаковыми аргументами:

from functools import lru_cache
import time

@lru_cache(maxsize=128)  # ✅ ОЧЕНЬ ЭФФЕКТИВНО
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Без кеша: ~5 сек
# С кешем: ~0.00001 сек (500,000x быстрее!)
print(fibonacci(40))  # Instant

В production используй Redis:

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_user_with_cache(user_id):
    cache_key = f"user:{user_id}"
    
    # Пытаемся получить из кеша
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Если нет — получаем из БД
    user = db.query(User).get(user_id)
    
    # Кешируем на 1 час
    redis_client.setex(cache_key, 3600, json.dumps(user))
    
    return user

5. Параллелизм: многопоточность vs многопроцессность

Threading для I/O-bound операций

import threading
import requests

def download_file(url):
    response = requests.get(url)
    return len(response.content)

urls = ["https://example.com/file1", "https://example.com/file2"] * 5

# ✅ Многопоточность (быстро для I/O)
threads = []
for url in urls:
    t = threading.Thread(target=download_file, args=(url,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()  # Ждём все потоки

Multiprocessing для CPU-bound операций

import multiprocessing

def cpu_intensive(n):
    result = 0
    for i in range(n):
        result += i ** 2
    return result

if __name__ == '__main__':
    # ✅ Многопроцессность (быстро для CPU)
    with multiprocessing.Pool(4) as pool:
        results = pool.map(cpu_intensive, [1000000] * 4)
    print(sum(results))

6. Асинхронное программирование (asyncio)

Мощный способ для высокой пропускной способности:

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def fetch_all_urls(urls):
    async with aiohttp.ClientSession() as session:
        # Все запросы параллельно!
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# 100 URL параллельно
results = asyncio.run(fetch_all_urls(["https://example.com"] * 100))
print(f"Загружено {len(results)} страниц")

7. Оптимизация работы с памятью

Утечки памяти — частая причина падения production-приложений:

import gc
import weakref

# ❌ Утечка памяти: циклические ссылки
class Node:
    def __init__(self, value):
        self.value = value
        self.next = None
        self.parent = None

# ✅ Исправление: используй weakref для обратных ссылок
class BetterNode:
    def __init__(self, value):
        self.value = value
        self.next = None
        self.parent = weakref.ref(None)  # Слабая ссылка

# Принудительная очистка
gc.collect()  # Запуск garbage collector

# Профилирование памяти
import tracemalloc
tracemalloc.start()

# ... код ...

current, peak = tracemalloc.get_traced_memory()
print(f"Текущая: {current / 1024 / 1024:.1f}MB, Пик: {peak / 1024 / 1024:.1f}MB")

8. Компилирование Python в C (Cython)

Для критичного кода используй Cython:

# fibonacci.pyx
cdef int fibonacci(int n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Компилируем в C
cythonize fibonacci.pyx
# Результат: в 50-100 раз быстрее

9. Оптимизация БД запросов

Запросы часто самые медленные части:

# ❌ N+1 проблема
for post in db.query(Post).all():
    print(post.author.name)  # Каждый проход = отдельный запрос

# ✅ Eager loading
from sqlalchemy.orm import joinedload
posts = db.query(Post).options(joinedload('author')).all()
for post in posts:
    print(post.author.name)  # Один запрос на всё

# ✅ Пакетная обработка
for post in db.query(Post).batch(100):  # По 100 строк
    process(post)

10. Практический пример: оптимизация API

from fastapi import FastAPI
from functools import lru_cache
import redis

app = FastAPI()
redis_client = redis.Redis()

@lru_cache(maxsize=1000)  # Кеш памяти
def get_user_by_id_cached(user_id: int):
    return f"User {user_id}"

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Многоуровневый кеш
    cache_key = f"user:{user_id}"
    
    # L1: Redis (быстро, разделяется между процессами)
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # L2: LRU (очень быстро, но локально)
    user = get_user_by_id_cached(user_id)
    
    # Кешируем в Redis
    redis_client.setex(cache_key, 3600, json.dumps(user))
    
    return user

Чеклист оптимизации (в порядке приоритета)

1. Профилируй — найди узкие места
2. Оптимизируй алгоритмы — Big O имеет значение
3. Кеширование — Redis, LRU, memcached
4. БД — индексы, eager loading, connection pooling
5. Асинхронность — asyncio для I/O
6. Многопроцессность — для CPU
7. Память — gc, weakref, профилирование
8. Компилирование — Cython как последняя линия защиты

Золотое правило

"Premature optimization is the root of all evil" — Donald Knuth. Но оптимизация медленного кода в production — это просто. Профилируй, анализируй, оптимизируй в таком порядке. Не гадай, измеряй.