#6 - Работа с БД + очереди

Асинхронная работа с очередями и базой данных — это один из базовых паттернов, который используется в реальных Python-приложениях. Когда программа получает данные (например, из API, файла или пользователя), не всегда удобно обрабатывать их сразу. Гораздо эффективнее разделить процесс на этапы: сначала получить данные, затем передать их через очередь и уже потом сохранить в базу. Такой подход делает код более структурированным и масштабируемым.

В основе этого подхода лежит модель producer/consumer. Producer отвечает за получение данных и добавление их в очередь, а consumer — за обработку этих данных. Между ними находится asyncio.Queue, которая работает асинхронно и не блокирует выполнение программы. Это значит, что пока одна часть кода ожидает данные, другая может продолжать работу. Такой механизм особенно важен при работе с сетью или файлами, где операции могут занимать время.

Для хранения данных используется база SQLite, но в асинхронном виде — через библиотеку aiosqlite. В отличие от обычного sqlite3, она позволяет выполнять запросы с использованием await, не блокируя event loop. Это критично, потому что любая блокирующая операция может “заморозить” всё асинхронное приложение.

Вся логика строится как простой pipeline: данные поступают в очередь, затем извлекаются и сохраняются в базу. При этом важно понимать, что очередь — это буфер между этапами. Она позволяет не зависеть от скорости выполнения каждой части системы. Например, данные могут поступать быстрее, чем сохраняются, и очередь временно их удерживает.

Ниже приведён минимальный пример, который показывает связку очереди и базы данных:

import asyncio
import aiosqlite

async def main():
	queue = asyncio.Queue()

	await queue.put({"name": "Keyboard", "price": 100})

	async with aiosqlite.connect("db.sqlite") as db:
		await db.execute("CREATE TABLE IF NOT EXISTS products (name TEXT, price REAL)")
 
		item = await queue.get()
		await db.execute("INSERT INTO products VALUES (?, ?)", (item["name"], item["price"]))
		await db.commit()

asyncio.run(main())

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