Есть ли критичные моменты в работе?

Критичные моменты в работе Python разработчика

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

1. Критичные моменты в базах данных

Race conditions и параллельный доступ

Когда несколько пользователей одновременно изменяют один ресурс, могут возникнуть непредсказуемые результаты.

# ПЛОХО - race condition
class BankAccount:
    def withdraw(self, amount: float):
        if self.balance >= amount:  # Проверка
            self.balance -= amount   # Изменение
            return True
        return False

# Два потока одновременно:
# Thread 1: проверка: balance=100, amount=100 ✓
# Thread 2: проверка: balance=100, amount=100 ✓
# Thread 1: вычитаем: balance = 0
# Thread 2: вычитаем: balance = -100 ← ОШИБКА!

# ХОРОШО - используем транзакции с блокировкой
from sqlalchemy.orm import Session
from sqlalchemy import select, and_

class BankAccount(Base):
    __tablename__ = "accounts"
    id = Column(Integer, primary_key=True)
    balance = Column(Float, default=0)

def withdraw_safely(account_id: int, amount: float, db: Session) -> bool:
    """Безопасное снятие со счёта"""
    try:
        # SELECT FOR UPDATE - другие транзакции ждут
        account = db.query(BankAccount).with_for_update().filter_by(
            id=account_id
        ).first()
        
        if account.balance >= amount:
            account.balance -= amount
            db.commit()
            return True
        
        db.rollback()
        return False
    except Exception as e:
        db.rollback()
        raise

Потеря данных при откатах

Если не обработать исключения, данные могут быть потеряны.

# ПЛОХО
def create_order(user_id: int, items: List[Item]):
    order = Order(user_id=user_id)
    db.add(order)
    db.flush()  # Получили ID
    
    for item in items:
        order_item = OrderItem(order_id=order.id, item_id=item.id)
        db.add(order_item)
        db.commit()  # Если ошибка на 5м товаре, 4 сохранены, 1-4 потеряны

# ХОРОШО - одна транзакция
def create_order(user_id: int, items: List[Item], db: Session):
    try:
        order = Order(user_id=user_id)
        db.add(order)
        db.flush()
        
        for item in items:
            order_item = OrderItem(order_id=order.id, item_id=item.id)
            db.add(order_item)
        
        db.commit()  # Либо все товары добавлены, либо ничего
        return order
    except Exception:
        db.rollback()
        raise

2. Критичные моменты безопасности

SQL Injection

Одна из самых опасных уязвимостей.

# КРИТИЧНО ОПАСНО - SQL Injection
user_input = "'; DROP TABLE users; --"
query = f"SELECT * FROM users WHERE email = '{user_input}'"  # БОМБА!
db.execute(query)

# БЕЗОПАСНО - параметризованные запросы
query = "SELECT * FROM users WHERE email = :email"
db.execute(query, {"email": user_input})  # Строка экранируется

# ИЛИ с SQLAlchemy ORM
from sqlalchemy import select
user = db.session.execute(
    select(User).where(User.email == user_input)
).scalar_one_or_none()  # Безопасно

Хранение паролей

Критично неправильное хранение паролей.

# КРИТИЧНО ОПАСНО
class User(Base):
    password = Column(String)  # Пароль в открытом виде!

user.password = user_input  # Сохраняем то, что прислал пользователь

# БЕЗОПАСНО
from werkzeug.security import generate_password_hash, check_password_hash

class User(Base):
    __tablename__ = "users"
    password_hash = Column(String, nullable=False)

def register_user(email: str, password: str, db: Session):
    user = User(
        email=email,
        password_hash=generate_password_hash(password)  # Хешируем
    )
    db.add(user)
    db.commit()

def verify_password(stored_hash: str, provided_password: str) -> bool:
    return check_password_hash(stored_hash, provided_password)

Экспозиция чувствительных данных

Данные могут утечь в логах, ошибках, кэшах.

# ОПАСНО
import logging

logger = logging.getLogger(__name__)

def process_payment(card_data: dict):
    logger.info(f"Processing payment: {card_data}")  # Логируем карточку!
    # Логи содержат чувствительные данные

# БЕЗОПАСНО
def process_payment(card_data: dict):
    # Логируем только необходимое
    logger.info(f"Processing payment for user, card last4: {card_data['number'][-4:]}")
    
    # Или используем маскирование
    def mask_sensitive(data):
        masked = data.copy()
        masked['card_number'] = "*" * 12 + masked['card_number'][-4:]
        return masked
    
    logger.info(f"Payment data: {mask_sensitive(card_data)}")

3. Критичные моменты в асинхронном коде

Deadlock'и и зависания

В асинхронном коде легко создать ситуацию, когда программа зависает.

# ОПАСНО - может зависнуть
import asyncio

async def fetch_data():
    async with aiohttp.ClientSession() as session:
        response = await session.get("https://api.example.com/data")  # Бесконечно
        return response

# БЕЗОПАСНО - с timeout
async def fetch_data():
    timeout = aiohttp.ClientTimeout(total=5)  # 5 секунд максимум
    async with aiohttp.ClientSession(timeout=timeout) as session:
        response = await session.get("https://api.example.com/data")
        return response

# ИЛИ
async def fetch_data_with_timeout():
    try:
        response = await asyncio.wait_for(
            fetch_data(),
            timeout=5.0
        )
        return response
    except asyncio.TimeoutError:
        logger.error("API request timed out")
        raise

Неправильное управление ресурсами

Асинхронные ресурсы (соединения, сессии) нужно закрывать правильно.

# ОПАСНО - утечка соединений
async def process_requests(urls: List[str]):
    session = aiohttp.ClientSession()
    tasks = [
        session.get(url) for url in urls
    ]
    results = await asyncio.gather(*tasks)
    # Забыли закрыть session! Утечка соединения
    return results

# БЕЗОПАСНО
async def process_requests(urls: List[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get(url) for url in urls
        ]
        results = await asyncio.gather(*tasks)
        # Сессия закроется автоматически
    return results

4. Критичные моменты в производительности

N+1 запросы

Максимально неэффективно, может убить производительность приложения.

# КРИТИЧНО ОПАСНО - N+1 запросы
users = db.query(User).all()  # 1 запрос

for user in users:  # Если пользователей 10 000
    posts = db.query(Post).filter_by(user_id=user.id).all()  # 10 000 запросов!
    # Итого: 10 001 запрос вместо 1!

# БЕЗОПАСНО - eager loading
from sqlalchemy.orm import joinedload

users = db.query(User).options(
    joinedload(User.posts)
).all()  # 1 запрос с JOIN

for user in users:
    posts = user.posts  # Данные уже загружены

Утечки памяти

От которых может упасть сервер в production.

# ОПАСНО - кэш растёт бесконечно
from functools import lru_cache

cache = {}  # Глобальный словарь

def process_file(filename: str):
    if filename not in cache:
        cache[filename] = read_large_file(filename)  # 100MB
    return cache[filename]

# После обработки 1000 файлов: 100GB памяти!

# БЕЗОПАСНО - ограниченный кэш
from functools import lru_cache

@lru_cache(maxsize=100)  # Максимум 100 записей
def process_file(filename: str):
    return read_large_file(filename)

# Или вручную с временем жизни
import time
from collections import OrderedDict

class TTLCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.cache = OrderedDict()
        self.ttl = ttl_seconds
        self.timestamps = {}
    
    def get(self, key: str):
        if key in self.cache:
            if time.time() - self.timestamps[key] < self.ttl:
                return self.cache[key]
            else:
                del self.cache[key]
                del self.timestamps[key]
        return None
    
    def set(self, key: str, value):
        self.cache[key] = value
        self.timestamps[key] = time.time()

5. Критичные моменты в API

Неправильная обработка ошибок

Может привести к утечке внутренней информации или плохому UX.

# ОПАСНО
@app.post("/login")
async def login(credentials: LoginRequest):
    user = db.query(User).filter_by(email=credentials.email).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")  # Утечка!
    if not verify_password(user.password, credentials.password):
        raise HTTPException(status_code=401, detail="Wrong password")  # Утечка!

# БЕЗОПАСНО - одинаковые ошибки
@app.post("/login")
async def login(credentials: LoginRequest):
    user = db.query(User).filter_by(email=credentials.email).first()
    if not user or not verify_password(user.password, credentials.password):
        raise HTTPException(
            status_code=401,
            detail="Invalid email or password"  # Одна ошибка для обоих случаев
        )
    return {"token": create_token(user.id)}

Отсутствие rate limiting

Открытые двери для DDoS и brute force.

# ОПАСНО - нет защиты
@app.post("/login")
async def login(credentials: LoginRequest):
    # Кто угодно может атаковать перебором
    ...

# БЕЗОПАСНО - rate limiting
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/login")
@limiter.limit("5/minute")  # 5 попыток в минуту
async def login(request: Request, credentials: LoginRequest):
    ...

6. Критичные моменты в развёртывании

Не проверены миграции в production

# ОПАСНО - миграция может сломать production
def migrate_user_emails():
    """На production это может быть критично!"""
    users = db.query(User).all()
    for user in users:
        # Что если это займёт 2 часа на 1M пользователей?
        # БД будет блокирована!
        user.email = user.email.lower()
    db.commit()

# БЕЗОПАСНО
def migrate_user_emails():
    """Батчами, с логированием, с возможностью отката"""
    batch_size = 1000
    offset = 0
    
    while True:
        users = db.query(User).offset(offset).limit(batch_size).all()
        if not users:
            break
        
        for user in users:
            user.email = user.email.lower()
        
        db.commit()
        logger.info(f"Migrated {offset + len(users)} users")
        offset += batch_size
        time.sleep(0.1)  # Снижаем нагрузку на БД

Чеклист критичных моментов

• ✓ Все HTTP запросы имеют timeout
• ✓ Все SQL запросы параметризованы
• ✓ Пароли хешируются (bcrypt/argon2)
• ✓ Чувствительные данные не логируются
• ✓ Race conditions обработаны (SELECT FOR UPDATE)
• ✓ Все ресурсы (файлы, соединения) закрываются
• ✓ Кэши имеют размер ограничения или TTL
• ✓ Нет N+1 запросов (eager loading)
• ✓ Есть rate limiting на публичных endpoints
• ✓ Миграции тестируются перед production
• ✓ Логируются ошибки, но не чувствительные данные
• ✓ Есть мониторинг и алерты в production

Вывод: Критичные моменты требуют тщательного внимания к деталям. Одна ошибка может привести к потере данных, утечке информации или краху сервиса. Всегда думайте о edge cases и тестируйте сценарии отказа.