aiopg

Асинхронная библиотека для работы с PostgreSQL, предоставляющая удобный интерфейс для управления подключениями, выполнения запросов и работы с репозиториями.

Возможности

• Пул соединений: Эффективное управление подключениями к PostgreSQL
• Асинхронные операции: Полная поддержка async/await для всех операций с базой данных
• JSON поддержка: Встроенная поддержка JSONB с оптимизированными кодерами
• Временные метки: Автоматическая обработка timestamp и timestamptz
• Репозитории: Готовые абстракции для работы с данными
• Система версионирования: Поддержка версионирования таблиц
• Обработка ошибок: Специализированные исключения для различных сценариев

Установка

pip install aiopg

Быстрый старт

Создание пула соединений

from aiopg import create_db_pool

# Создание пула соединений
pool = create_db_pool(
    dsn="postgresql://user:password@localhost/dbname",
    min_size=5,
    max_size=20
)

Выполнение запросов

from aiopg import compile_query

# Компиляция и выполнение запроса
query = compile_query("SELECT * FROM users WHERE id = $1")
async with pool.acquire() as conn:
    result = await conn.fetch(query, user_id)

Работа с репозиториями

from aiopg.repository import PGDataAccessObject, TableDescriptor

# Определение таблицы
users_table = TableDescriptor("users", ["id", "name", "email"])

# Создание DAO
users_dao = PGDataAccessObject(pool, users_table)

# Операции с данными
user = await users_dao.get_by_id(user_id)
users = await users_dao.get_all()
await users_dao.insert({"name": "John", "email": "john@example.com"})

Транзакции

PostgresAccessLayer.transaction() объединяет вызовы всех DAO этого слоя в одну транзакцию на одном соединении — включая кастомные методы DAO, использующие fetch/fetchrow/fetchval. Commit при выходе из блока, rollback при исключении:

async with dal.transaction():
    wish = await dal.wishes.create(owner_id=user.id, **payload)
    await dal.wishlist_items.create(wishlist_id=default.id, wish_id=wish.id)

Вне блока поведение прежнее: каждый вызов берёт своё соединение из пула (autocommit).

• Вложенные transaction() на том же пуле переиспользуют соединение и открывают savepoint: откат внутреннего блока не отменяет внешнюю транзакцию. ⚠️ До 0.3.0 вложенный вызов брал отдельное соединение с независимой транзакцией.
• Несколько пулов в одном процессе изолированы: транзакция одного слоя не затрагивает запросы слоёв с другими пулами.
• asyncio.gather внутри блока не поддерживается: соединение asyncpg нельзя использовать конкурентно, будет InterfaceError: another operation is in progress. Выполняйте вызовы последовательно.
• asyncio.create_task внутри блока: таск наследует транзакционный контекст и обязан завершиться до выхода из блока, иначе он может обратиться к уже возвращённому в пул соединению.

Основные компоненты

Connection Pool

• create_db_pool() - создание пула соединений с автоматической инициализацией кодеков

Query Compilation

• compile_query() - компиляция SQL запросов для оптимизации

Repository Pattern

• PGDataAccessObject - базовый DAO для работы с таблицами
• PostgresAccessLayer - слой доступа к PostgreSQL
• PGPoolManager - менеджер пулов соединений

Version Management

• declare_version_table() - создание таблиц для версионирования

Требования

• Python >= 3.8
• PostgreSQL
• asyncpg >= 0.27.0
• psycopg2-binary >= 2.9.0

Лицензия

MIT License - см. файл LICENSE для подробностей.

Разработка

# Клонирование репозитория
git clone https://github.com/ascet-dev/adc-aiopg.git
cd adc-aiopg

# Установка зависимостей для разработки
pip install -e .