Ускорение Python: как asyncio повышает производительность приложений

import asyncio

async def hello_world():
print("Hello")
await asyncio.sleep(1) # Имитация I/O операции
print("World")

asyncio.run(hello_world())

# Создание виртуального окружения
python -m venv asyncio_env

# Активация в Linux/macOS
source asyncio_env/bin/activate

# Активация в Windows
# asyncio_env\Scripts\activate

# Установка базовых зависимостей
pip install aiohttp aiofiles pytest-asyncio

import logging
import sys

logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)

# Получение логгера для asyncio
asyncio_logger = logging.getLogger('asyncio')

project/
├── app/
│ ├── __init__.py
│ ├── client.py # Асинхронный HTTP-клиент
│ ├── processor.py # Обработчик данных
│ └── storage.py # Работа с БД или файлами
├── config.py # Конфигурация приложения
└── main.py # Точка входа

# client.py
import aiohttp
import asyncio
import logging
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)

class APIClient:
def __init__(self, base_url: str, timeout: int = 30):
self.base_url = base_url
self.timeout = aiohttp.ClientTimeout(total=timeout)
self._session: Optional[aiohttp.ClientSession] = None

async def __aenter__(self):
if self._session is None:
self._session = aiohttp.ClientSession(
timeout=self.timeout,
raise_for_status=True
)
return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._session:
await self._session.close()
self._session = None

async def get(self, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
if not self._session:
raise RuntimeError("Client session is not initialized. Use 'async with' pattern.")

url = f"{self.base_url}/{endpoint}"
logger.debug(f"GET request to {url} with params {params}")

try:
async with self._session.get(url, params=params) as response:
return await response.json()
except aiohttp.ClientError as e:
logger.error(f"Error during GET request to {url}: {e}")
raise

# processor.py
import asyncio
import logging
from typing import List, Dict, Any

logger = logging.getLogger(__name__)

class DataProcessor:
def __init__(self, concurrency_limit: int = 10):
self.semaphore = asyncio.Semaphore(concurrency_limit)

async def process_item(self, item: Dict[str, Any]) -> Dict[str, Any]:
# Имитация сложной обработки
logger.debug(f"Processing item: {item['id']}")
await asyncio.sleep(0.5) # Имитация I/O операции

# Трансформация данных
result = {
"processed_id": item["id"],
"title_length": len(item.get("title", "")),
"is_completed": item.get("completed", False),
"timestamp": asyncio.current_task().get_name()
}
logger.debug(f"Processed item {item['id']}")
return result

async def process_batch(self, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
async def _process_with_semaphore(item):
async with self.semaphore:
return await self.process_item(item)

tasks = [asyncio.create_task(
_process_with_semaphore(item),
name=f"process-{item['id']}"
) for item in items]

results = await asyncio.gather(*tasks, return_exceptions=True)

# Фильтрация результатов, исключение ошибок
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"Error processing item {items[i]['id']}: {result}")
else:
processed_results.append(result)

return processed_results

# storage.py
import aiofiles
import json
import logging
from typing import List, Dict, Any

logger = logging.getLogger(__name__)

class AsyncStorage:
def __init__(self, file_path: str):
self.file_path = file_path

async def save_results(self, results: List[Dict[str, Any]]) -> None:
logger.info(f"Saving {len(results)} results to {self.file_path}")

async with aiofiles.open(self.file_path, 'w') as f:
# Преобразование в JSON и сохранение
json_str = json.dumps(results, indent=2)
await f.write(json_str)

logger.info(f"Successfully saved results to {self.file_path}")

async def load_results(self) -> List[Dict[str, Any]]:
try:
async with aiofiles.open(self.file_path, 'r') as f:
content = await f.read()
return json.loads(content)
except FileNotFoundError:
logger.warning(f"File {self.file_path} not found, returning empty list")
return []

# main.py
import asyncio
import logging
from app.client import APIClient
from app.processor import DataProcessor
from app.storage import AsyncStorage
from config import settings

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def main():
# Инициализация компонентов
client = APIClient(base_url=settings.API_BASE_URL)
processor = DataProcessor(concurrency_limit=settings.CONCURRENCY_LIMIT)
storage = AsyncStorage(file_path=settings.RESULTS_FILE)

async with client:
# Получение данных
data = await client.get("todos")

# Обработка данных
processed_data = await processor.process_batch(data)

# Сохранение результатов
await storage.save_results(processed_data)

logger.info("Application completed successfully")

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

# Неэффективно: создание тысяч отдельных задач
tasks = [asyncio.create_task(process_item(item)) for item in items]
results = await asyncio.gather(*tasks)

# Эффективно: обработка данных батчами
batch_size = 100
results = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
batch_results = await process_batch(batch)
results.extend(batch_results)

# Оптимизированный HTTP-клиент с пулом соединений
class OptimizedAPIClient:
def __init__(self, base_url, pool_size=100, ssl=None):
self.base_url = base_url
self.connector = aiohttp.TCPConnector(limit=pool_size, ssl=ssl)
self.session = None

async def __aenter__(self):
self.session = aiohttp.ClientSession(
connector=self.connector,
timeout=aiohttp.ClientTimeout(total=60),
raise_for_status=True
)
return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()

# Реализация методов get/post/etc.

async def process_data_stream(source, sink, max_concurrent=10):
semaphore = asyncio.Semaphore(max_concurrent)
queue = asyncio.Queue()

# Продюсер – получает данные из источника
async def producer():
async for item in source:
await queue.put(item)
# Сигнал завершения
await queue.put(None)

# Потребитель – обрабатывает данные и отправляет в приемник
async def consumer():
while True:
item = await queue.get()
if item is None: # Сигнал завершения
queue.task_done()
break

async with semaphore:
processed_item = await process_item(item)
await sink.send(processed_item)

queue.task_done()

# Запускаем продюсера и несколько потребителей
producer_task = asyncio.create_task(producer())
consumer_tasks = [
asyncio.create_task(consumer())
for _ in range(max_concurrent)
]

# Ждем завершения всех задач
await producer_task
await queue.join()
await asyncio.gather(*consumer_tasks)

async def measure(coro, name=None):
start_time = time.time()
try:
return await coro
finally:
execution_time = time.time() – start_time
logging.info(f"Operation {name or coro.__name__} took {execution_time:.4f} seconds")
# Здесь можно отправить метрику в систему мониторинга

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import logging
import time
from urllib.parse import urljoin
from typing import List, Dict, Set

class AsyncWebScraper:
def __init__(self, concurrency=10, rate_limit=2):
self.concurrency = concurrency
self.rate_limit = 1/rate_limit # секунды между запросами
self.visited_urls: Set[str] = set()
self.semaphore = asyncio.Semaphore(concurrency)
self.last_request_time = 0

async def throttle(self):
"""Ограничение частоты запросов"""
now = time.time()
elapsed = now – self.last_request_time
if elapsed < self.rate_limit:
await asyncio.sleep(self.rate_limit – elapsed)
self.last_request_time = time.time()

async def fetch_page(self, session, url):
"""Получение страницы с соблюдением ограничений"""
if url in self.visited_urls:
return None

self.visited_urls.add(url)

async with self.semaphore:
await self.throttle()
try:
async with session.get(url) as response:
if response.status != 200:
logging.warning(f"Status {response.status} for {url}")
return None
return await response.text()
except Exception as e:
logging.error(f"Error fetching {url}: {e}")
return None

async def parse_page(self, html, url):
"""Извлечение данных из HTML"""
if not html:
return None

soup = BeautifulSoup(html, 'html.parser')

# Пример извлечения данных (настройте под свои нужды)
title = soup.title.text.strip() if soup.title else "No title"
paragraphs = [p.text for p in soup.find_all('p')]
links = [urljoin(url, a['href']) for a in soup.find_all('a', href=True)]

return {
"url": url,
"title": title,
"text_length": len(' '.join(paragraphs)),
"links": links
}

async def scrape_url_with_depth(self, start_url, max_depth=2):
"""Рекурсивный скрапинг с ограничением глубины"""
results = []
queue = [(start_url, 0)] # (url, depth)

async with aiohttp.ClientSession() as session:
while queue:
url, depth = queue.pop(0)

if depth > max_depth:
continue

html = await self.fetch_page(session, url)
data = await self.parse_page(html, url)

if data:
results.append(data)
# Добавляем новые ссылки в очередь
if depth < max_depth:
for link in data["links"][:5]: # Ограничиваем количество ссылок
if link not in self.visited_urls:
queue.append((link, depth + 1))

return results

# Использование:
async def main():
scraper = AsyncWebScraper(concurrency=5, rate_limit=1)
results = await scraper.scrape_url_with_depth("https://example.com", max_depth=2)
print(f"Scraped {len(results)} pages")

asyncio.run(main())

from aiohttp import web
import asyncio
import logging
import json
from typing import Dict, List, Any
import aioredis

# Имитация хранилища данных
class AsyncDataProcessor:
def __init__(self):
self.redis = None

async def connect(self):
self.redis = await aioredis.create_redis_pool('redis://localhost')

async def close(self):
if self.redis:
self.redis.close()
await self.redis.wait_closed()

async def process_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
# Имитация сложной обработки
await asyncio.sleep(0.5)

# Сохранение в Redis
key = f"data:{data.get('id', 'unknown')}"
await self.redis.set(key, json.dumps(data))

return {
"status": "processed",
"id": data.get("id"),
"timestamp": time.time()
}

async def get_stats(self) -> Dict[str, Any]:
# Получение статистики
keys = await self.redis.keys("data:*")
count = len(keys)

return {
"total_processed": count,
"last_updated": time.time()
}

# Настройка приложения aiohttp
async def create_app():
app = web.Application()
processor = AsyncDataProcessor()

# Регистрация процессора в приложении
app['processor'] = processor

async def on_startup(app):
await app['processor'].connect()

async def on_shutdown(app):
await app['processor'].close()

app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)

# Обработчики API
async def handle_process(request):
try:
data = await request.json()
except json.JSONDecodeError:
return web.json_response({"error": "Invalid JSON"}, status=400)

result = await request.app['processor'].process_data(data)
return web.json_response(result)

async def handle_stats(request):
stats = await request.app['processor'].get_stats()
return web.json_response(stats)

# Регистрация маршрутов
app.router.add_post('/api/process', handle_process)
app.router.add_get('/api/stats', handle_stats)

return app

# Запуск приложения
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
app = asyncio.run(create_app())
web.run_app(app, port=8080)

import asyncio
import json
import logging
from typing import Callable, Dict, Any
import aio_pika

class AsyncMessageProcessor:
def __init__(self, connection_url: str, queue_name: str, max_workers: int = 5):
self.connection_url = connection_url
self.queue_name = queue_name
self.max_workers = max_workers
self.connection = None
self.channel = None
self.handlers = {}

async def connect(self):
self.connection = await aio_pika.connect_robust(self.connection_url)
self.channel = await self.connection.channel()
await self.channel.set_qos(prefetch_count=self.max_workers)

# Объявление очереди
self.queue = await self.channel.declare_queue(
self.queue_name,
durable=True
)

async def close(self):
if self.connection:
await self.connection.close()

def register_handler(self, message_type: str, handler: Callable):
"""Регистрация обработчика для определенного типа сообщений"""
self.handlers[message_type] = handler

async def process_message(self, message):
async with message.process():
try:
body = message.body.decode()
data = json.loads(body)

# Определение типа сообщения
message_type = data.get("type")
if not message_type:
logging.error(f"Message has no type field: {data}")
return

# Поиск и вызов соответствующего обработчика
handler = self.handlers.get(message_type)
if not handler:
logging.warning(f"No handler for message type: {message_type}")
return

# Обработка сообщения
await handler(data)
logging.info(f"Successfully processed {message_type} message")

except json.JSONDecodeError:
logging.error(f"Failed to parse message JSON: {message.body}")
except Exception as e:
logging.exception(f"Error processing message: {e}")

async def start_consuming(self):
"""Начать потребление сообщений из очереди"""
logging.info(f"Starting to consume messages from {self.queue_name}")
await self.queue.consume(self.process_message)

# Держим соединение открытым
try:
await asyncio.Future()
except asyncio.CancelledError:
logging.info("Consumer was cancelled, shutting down")

# Пример использования:
async def example_handler(data):
logging.info(f"Processing data: {data}")
await asyncio.sleep(1) # Имитация работы
return {"status": "completed"}

async def main():
processor = AsyncMessageProcessor(
connection_url="amqp://guest:guest@localhost/",
queue_name="tasks"
)

# Регистрация обработчиков
processor.register_handler("task.create", example_handler)
processor.register_handler("task.update", example_handler)

await processor.connect()
try:
await processor.start_consuming()
finally:
await processor.close()

if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(main())