Приведи пример трёх интересных задач

Три интересные задачи Python Developer

Вот примеры задач, которые я считаю интересными и часто встречаю в реальных проектах:

1. Оптимизация работы с большими данными через асинхронность

Задача: разработать систему, которая загружает 10000+ записей из API, обрабатывает их, и сохраняет в БД. Наивный подход (последовательные запросы) займёт часы.

Решение использует асинхронность и пулинг соединений:

import asyncio
import aiohttp
from typing import List
from concurrent.futures import ThreadPoolExecutor

async def fetch_data(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        return await resp.json()

async def process_batch(urls: List[str]) -> List[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

def save_to_db(data: List[dict]) -> None:
    # Использование ThreadPoolExecutor для блокирующих операций БД
    with ThreadPoolExecutor(max_workers=5) as executor:
        loop = asyncio.get_event_loop()
        loop.run_in_executor(executor, bulk_insert, data)

async def main():
    urls = [f"https://api.example.com/items/{i}" for i in range(10000)]
    batches = [urls[i:i+100] for i in range(0, len(urls), 100)]
    
    for batch in batches:
        results = await process_batch(batch)
        save_to_db(results)

Что делает эту задачу интересной: управление асинхронностью, обработка ошибок, пулинг ресурсов, масштабируемость.

2. Разработка микросервиса с кешированием и инвалидацией

Задача: создать API сервис для подсчёта статистики пользователей. Данные дорогие (требуют агрегации 100k+ записей), но запрашиваются часто.

Решение: многоуровневое кеширование с правильной инвалидацией:

from functools import wraps
import hashlib
import json
from datetime import datetime, timedelta
from typing import Callable, Any
import redis
from fastapi import FastAPI, HTTPException
from sqlalchemy import select, func
from database import async_session, User, Activity

app = FastAPI()
redis_client = redis.Redis(host="localhost", decode_responses=True)

class CacheManager:
    def __init__(self, ttl: int = 3600):
        self.ttl = ttl
    
    def get_key(self, func_name: str, args: tuple, kwargs: dict) -> str:
        data = f"{func_name}:{json.dumps((args, kwargs), sort_keys=True, default=str)}"
        return hashlib.md5(data.encode()).hexdigest()
    
    def cached(self, ttl: int = None):
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            async def wrapper(*args, **kwargs) -> Any:
                cache_ttl = ttl or self.ttl
                cache_key = self.get_key(func.__name__, args, kwargs)
                
                # Попытка получить из кеша
                cached_value = redis_client.get(cache_key)
                if cached_value:
                    return json.loads(cached_value)
                
                # Вычисляем значение
                result = await func(*args, **kwargs)
                redis_client.setex(
                    cache_key,
                    cache_ttl,
                    json.dumps(result, default=str)
                )
                return result
            return wrapper
        return decorator
    
    def invalidate(self, pattern: str = "*") -> int:
        """Инвалидирует все ключи по паттерну"""
        keys = redis_client.keys(pattern)
        if keys:
            return redis_client.delete(*keys)
        return 0

cache = CacheManager(ttl=1800)

@app.get("/users/{user_id}/stats")
@cache.cached(ttl=600)
async def get_user_stats(user_id: int):
    async with async_session() as session:
        user = await session.execute(
            select(User).where(User.id == user_id)
        )
        if not user.scalar():
            raise HTTPException(status_code=404)
        
        activity_count = await session.execute(
            select(func.count(Activity.id)).where(Activity.user_id == user_id)
        )
        
        return {
            "user_id": user_id,
            "activities": activity_count.scalar(),
            "cached_at": datetime.utcnow().isoformat()
        }

@app.post("/users/{user_id}/activity")
async def create_activity(user_id: int):
    # ... создание активности
    # Инвалидируем кеш для этого пользователя
    cache.invalidate(f"get_user_stats*{user_id}*")
    return {"status": "created"}

Интересность: кеширование, инвалидация, паттерны проектирования, работа с Redis.

3. Реализация сложного алгоритма обработки данных (DFS для графов)

Задача: найти все циклы в ориентированном графе и вычислить сильно связанные компоненты (SCC).

Это типичная задача на системное проектирование, социальные сети, рекомендации.

from typing import Dict, List, Set, Tuple
from collections import defaultdict

class GraphAnalyzer:
    def __init__(self):
        self.graph: Dict[int, List[int]] = defaultdict(list)
        self.visited: Set[int] = set()
        self.rec_stack: Set[int] = set()  # Current recursion stack
        self.cycles: List[List[int]] = []
    
    def add_edge(self, u: int, v: int) -> None:
        self.graph[u].append(v)
    
    def find_cycles_dfs(self, node: int, path: List[int]) -> None:
        """Находит все циклы через DFS"""
        self.visited.add(node)
        self.rec_stack.add(node)
        path.append(node)
        
        for neighbor in self.graph[node]:
            if neighbor not in self.visited:
                self.find_cycles_dfs(neighbor, path.copy())
            elif neighbor in self.rec_stack:
                # Найден цикл
                cycle_start_idx = path.index(neighbor)
                cycle = path[cycle_start_idx:] + [neighbor]
                if cycle not in self.cycles:
                    self.cycles.append(cycle)
        
        self.rec_stack.remove(node)
    
    def find_all_cycles(self) -> List[List[int]]:
        for node in self.graph:
            if node not in self.visited:
                self.find_cycles_dfs(node, [])
        return self.cycles
    
    def tarjan_scc(self) -> List[List[int]]:
        """Находит сильно связанные компоненты (Tarjan алгоритм)"""
        self.index_counter = 0
        self.stack: List[int] = []
        self.lowlinks: Dict[int, int] = {}
        self.index: Dict[int, int] = {}
        self.on_stack: Set[int] = set()
        self.sccs: List[List[int]] = []
        
        for node in self.graph:
            if node not in self.index:
                self._strongconnect(node)
        
        return self.sccs
    
    def _strongconnect(self, node: int) -> None:
        self.index[node] = self.index_counter
        self.lowlinks[node] = self.index_counter
        self.index_counter += 1
        self.stack.append(node)
        self.on_stack.add(node)
        
        for neighbor in self.graph[node]:
            if neighbor not in self.index:
                self._strongconnect(neighbor)
                self.lowlinks[node] = min(self.lowlinks[node], self.lowlinks[neighbor])
            elif neighbor in self.on_stack:
                self.lowlinks[node] = min(self.lowlinks[node], self.index[neighbor])
        
        if self.lowlinks[node] == self.index[node]:
            scc = []
            while True:
                w = self.stack.pop()
                self.on_stack.remove(w)
                scc.append(w)
                if w == node:
                    break
            self.sccs.append(scc)

# Пример использования
analyzer = GraphAnalyzer()
for u, v in [(1, 2), (2, 3), (3, 1), (3, 4), (4, 5), (5, 3)]:
    analyzer.add_edge(u, v)

print("Циклы:", analyzer.find_all_cycles())
print("SCC:", analyzer.tarjan_scc())

Что делает интересной: алгоритмы графов, оптимизация, рекурсия, сложность O(V+E).

Почему эти задачи интересны

1. Практическое применение — встречаются в реальных проектах
2. Требуют глубокого понимания — асинхронность, кеширование, алгоритмы
3. Scalability — нужно думать о производительности
4. Комбинируют знания — архитектура, паттерны, алгоритмы

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