Реализация сложных Python проектов с продвинутыми алгоритмами и архитектурой

Расширенные примеры кода для сложных Python программ

Асинхронный парсинг с обработкой ошибок и таймером

Следующий пример демонстрирует загрузку 50 страниц с замером времени и логированием ошибок. Используется получевие и семафор для ограничения параллельных запросов.

import asyncio
import aiohttp
import time

async def fetch(session, url, semaphore):
    async with semaphore:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                return await resp.text()
        except Exception as e:
            return f"Ошибка {url}: {e}"

async def main():
    urls = [f"https://httpbin.org/delay/1?n={i}" for i in range(50)]
    sem = asyncio.Semaphore(5)  # не более 5 одновременных запросов
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, sem) for url in urls]
        start = time.time()
        results = await asyncio.gather(*tasks)
        elapsed = time.time() - start
    print(f"Время: {elapsed:.2f} сек")
    errors = [r for r in results if r.startswith("Ошибка")]
    print(f"Ошибок: {len(errors)}")

if __name__ == "__main__":
    asyncio.run(main())

Время: 10.12 сек
Ошибок: 0

Благодаря семафору не создаётся одновременная нагрузка в 50 запросов, что снижает риск блокировки.

Многопоточное скачивание с прогрессом через tqdm

Пример использует concurrent.futures.ThreadPoolExecutor и библиотеку tqdm для отображения прогресса.

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm

urls = ["https://httpbin.org/get"] * 30

def download(url):
    resp = requests.get(url, timeout=5)
    return (url, len(resp.text))

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(download, url): url for url in urls}
    for future in tqdm(as_completed(futures), total=len(urls), desc="Загрузка"):
        url, length = future.result()
        # можно сохранять результат

Загрузка: 100%|██████████| 30/30 [00:03<00:00,  8.99it/s]

Асинхронное обновление прогресс-бара происходит благодаря tqdm, которое перерисовывает строку при каждом завершении задачи.

Многопроцессорная обработка изображений с Pool

Реальный случай: преобразование 1000 изображений (уменьшение размера) с задействованием всех ядер CPU.

import os
import sys
from PIL import Image
from multiprocessing import Pool

def resize_image(filename):
    img = Image.open(filename)
    img = img.resize((100, 100))
    out_name = f"thumb_{filename}"
    img.save(out_name)
    return out_name

if __name__ == "__main__":
    files = [f"img_{i}.jpg" for i in range(1000)]
    with Pool(processes=os.cpu_count()) as pool:
        results = pool.map(resize_image, files)
    print(f"Создано {len(results)} миниатюр")

Создано 1000 миниатюр

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

Кастомный декоратор для измерения времени выполнения

Использование декораторов - один из признаков сложной программы. Пример автоматически замеряет время работы любой функции и выводит метрики.

import functools
import time

def log_time(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} выполнена за {elapsed:.4f} сек")
        return result
    return wrapper

@log_time
def heavy_calculation(n):
    return sum(i**2 for i in range(n))

print(heavy_calculation(10_000_000))

heavy_calculation выполнена за 0.4712 сек
333333283333335000000

Декоратор использован без изменения исходной функции. При добавлении stacklevel в wraps сохраняется сигнатура.

Контекстный менеджер для временной смены рабочей директории

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

import os
from contextlib import contextmanager

@contextmanager
def changedir(path):
    old_dir = os.getcwd()
    try:
        os.chdir(path)
        yield
    finally:
        os.chdir(old_dir)

with changedir("/tmp"):
    print("Текущая директория:", os.getcwd())
print("После выхода:", os.getcwd())

Текущая директория: /tmp
После выхода: /home/user

Реализация через contextmanager упрощает создание своих менеджеров, избегая класса с __enter__ и __exit__.