Как работаешь с рисками при выполнении задач?

Работа с рисками при выполнении задач

Это важный вопрос о профессиональном подходе к разработке. Я практикую систематическое управление рисками на всех этапах разработки.

1. Идентификация рисков на этапе планирования

Технические риски:

# Пример: выполнение задачи по внедрению нового подхода

# РИСК 1: Неизвестная библиотека — может быть неконсистентна
# РЕШЕНИЕ: написать POC (proof of concept)
from unknown_library import new_approach

try:
    # Тестируем в изолированной среде
    result = new_approach.test_basic()
    result = new_approach.test_edge_case()
except Exception as e:
    # Если риск реализовался — выбираем альтернативу
    # Используем проверенную библиотеку вместо неизвестной
    pass

Риск производительности:

# Перед написанием кода
# РИСК: может быть медленным на больших данных
# РЕШЕНИЕ: сначала написать бенчмарк

import time
from typing import List

def naive_approach(items: List[int]) -> List[int]:
    """O(n²) — медленно для больших данных"""
    result = []
    for item in items:
        if item not in result:  # O(n) проверка
            result.append(item)
    return result

def optimized_approach(items: List[int]) -> List[int]:
    """O(n) — оптимально"""
    return list(dict.fromkeys(items))  # Сохраняет порядок

# Бенчмарк
data = list(range(10000)) * 10

start = time.time()
naive_approach(data)
naive_time = time.time() - start

start = time.time()
optimized_approach(data)
opt_time = time.time() - start

print(f"Naive: {naive_time:.4f}s, Optimized: {opt_time:.4f}s")
# Результат: показывает, какой подход выбрать

Риск совместимости:

# РИСК: новая версия библиотеки несовместима
# РЕШЕНИЕ: явно указывать версию

# requirements.txt
django==4.2.0  # Фиксируем версию, не >=
celery==5.3.0
redis==5.0.0

# Или использовать поддиапазон для безопасности
# django>=4.2.0,<5.0.0  # Совместимо с 4.x

2. Проверка рисков через тесты

Unit тесты для граничных случаев:

import pytest

class UserService:
    def calculate_discount(self, age: int) -> float:
        """РИСК: неправильное поведение на граничных значениях"""
        if age < 18:
            return 0.2  # 20% скидка для молодёжи
        elif age >= 65:
            return 0.3  # 30% скидка для пенсионеров
        return 0.0

class TestUserService:
    def setup_method(self):
        self.service = UserService()
    
    # Граничные случаи
    def test_age_zero(self):
        assert self.service.calculate_discount(0) == 0.2
    
    def test_age_seventeen(self):
        assert self.service.calculate_discount(17) == 0.2
    
    def test_age_eighteen(self):
        assert self.service.calculate_discount(18) == 0.0
    
    def test_age_sixty_four(self):
        assert self.service.calculate_discount(64) == 0.0
    
    def test_age_sixty_five(self):
        assert self.service.calculate_discount(65) == 0.3
    
    def test_negative_age(self):
        """РИСК: отрицательный возраст"""
        with pytest.raises(ValueError):
            self.service.calculate_discount(-5)
    
    def test_very_large_age(self):
        """РИСК: чрезвычайно большое число"""
        assert self.service.calculate_discount(999) == 0.3

Интеграционные тесты для сложного взаимодействия:

# РИСК: БД может быть недоступна
# РЕШЕНИЕ: обработка исключений и retry logic

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def connect_to_database():
    """Автоматически переподключится до 3 раз"""
    try:
        db = DatabaseConnection()
        return db
    except ConnectionError:
        # Первая попытка — ждём 2 сек
        # Вторая попытка — ждём 4 сек
        # Третья попытка — ждём 8 сек
        # Если всё ещё не работает — поднимаем исключение
        raise

3. Обработка ошибок в production

Правильная обработка исключений:

from typing import Optional
from logging import getLogger

logger = getLogger(__name__)

class OrderService:
    def process_payment(self, order_id: str) -> Optional[str]:
        """РИСК: платёж может не пройти, сервис может быть недоступен"""
        try:
            payment_result = self.payment_gateway.charge(order_id)
            return payment_result.transaction_id
        
        except PaymentGatewayUnavailable:
            # Специфичное исключение — знаем, что делать
            logger.warning(f"Payment gateway unavailable for order {order_id}")
            # Пересчитываем запрос позже
            self.queue_for_retry(order_id)
            return None
        
        except InsufficientFunds:
            # Бизнес-ошибка — логируем и сообщаем пользователю
            logger.info(f"Insufficient funds for order {order_id}")
            raise OrderPaymentFailed("Your card has insufficient funds")
        
        except Exception as e:
            # Неожиданное исключение — логируем и alert
            logger.error(
                f"Unexpected error processing payment for {order_id}",
                exc_info=True  # Полная информация об ошибке
            )
            # Отправляем alert команде
            self.send_alert(f"Unknown payment error: {type(e).__name__}")
            raise SystemError("Payment processing failed unexpectedly")

4. Мониторинг и логирование рисков

Правильная структурированная логика:

import logging
from pythonjsonlogger import jsonlogger

# JSON логи для лучшей индексации в ELK/DataDog
logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)

class RiskMonitor:
    def log_slow_query(self, query: str, duration: float):
        """РИСК: медленные запросы могут замедлить приложение"""
        if duration > 1.0:  # Если дольше 1 сек
            logger.warning(
                "Slow database query detected",
                extra={
                    "query": query[:100],  # Первые 100 символов
                    "duration_ms": int(duration * 1000),
                    "severity": "medium" if duration < 5 else "high"
                }
            )
            # На основе логов можно создать alert в мониторинге

5. Code Review для выявления рисков

Чек-лист при review:

# ✗ РИСК 1: SQL injection
# Плохо
query = f"SELECT * FROM users WHERE id = {user_id}"

# Хорошо
query = "SELECT * FROM users WHERE id = %s"
results = db.execute(query, (user_id,))

# ✗ РИСК 2: Race condition
# Плохо
if user.balance >= amount:  # Проверка
    user.balance -= amount   # Операция
    user.save()             # Может быть несогласованность

# Хорошо
with db.transaction():
    user = User.objects.select_for_update().get(id=user_id)
    if user.balance >= amount:
        user.balance -= amount
        user.save()

# ✗ РИСК 3: Утечка памяти
# Плохо
global_cache = {}
def cache_user(user_id):
    global_cache[user_id] = get_user(user_id)  # Растёт бесконечно

# Хорошо
from functools import lru_cache

@lru_cache(maxsize=1000)  # Ограничиваем размер
def cache_user(user_id):
    return get_user(user_id)

# ✗ РИСК 4: Секреты в коде
# Плохо
api_key = "sk_live_51234567890"
db_password = "super_secret_123"

# Хорошо
from os import getenv
api_key = getenv('STRIPE_API_KEY')
db_password = getenv('DB_PASSWORD')

6. Уменьшение рисков через архитектуру

Разделение ответственности:

# РИСК: сложная функция с многими зависимостями
# Решение: разбить на меньшие, тестируемые компоненты

class PaymentProcessor:
    def __init__(
        self,
        validator: PaymentValidator,
        gateway: PaymentGateway,
        repository: OrderRepository,
        logger: Logger
    ):
        self.validator = validator
        self.gateway = gateway
        self.repository = repository
        self.logger = logger
    
    def process(self, order_id: str) -> bool:
        # Каждый шаг можно тестировать отдельно
        order = self.repository.get(order_id)
        
        if not self.validator.validate(order):
            raise ValidationError("Order invalid")
        
        result = self.gateway.charge(order.amount)
        self.repository.update_payment_status(order_id, result.status)
        self.logger.info(f"Payment processed for order {order_id}")
        
        return result.success

7. Общий подход к рискам

1. Планирование — Выявляю риски ДО написания кода
2. Testing — Пишу тесты для рисков
3. Code Review — Проверяю на типовые ошибки
4. Monitoring — Отслеживаю проблемы в production
5. Эскалация — Быстро сообщаю о критичных проблемах

Главный принцип: Лучше потратить час на выявление риска, чем день на отладку production инцидента.

В работе над задачами я подхожу прагматично:

• Для критичных задач — детальный анализ рисков
• Для рутинных — стандартные проверки
• Всегда имею fallback план
• Готов быстро масштабировать обработку ошибок