dbalchemycore 0.1.4

Asynchronous database library using SQLAlchemy Core with CRUD operations

dbalchemycore

dbalchemycore — это асинхронная библиотека для работы с базой данных PostgreSQL через SQLAlchemy Core. Она предоставляет удобный интерфейс для выполнения CRUD-операций, управления подключением к базе данных, миграциями через Alembic и асинхронными сессиями. Библиотека ориентирована на производительность, минимизацию дублирования кода и интеграцию с FastAPI и другими асинхронными фреймворками.

Основные возможности

• Асинхронные CRUD-операции: Поддержка создания, чтения, обновления и удаления записей через обобщённый класс BaseRepository.
• Pydantic-валидация: Использование Pydantic-схем для фильтров, значений и агрегаций.
• Поддержка миграций: Интеграция с Alembic для управления схемой базы данных.
• Управление транзакциями: Декоратор connection() для автоматического управления асинхронными сессиями SQLAlchemy.
• Гибкость: Поддержка фильтрации, сортировки, пагинации, группировки и агрегаций (count, sum, avg, min, max).
• Логирование: Встроенное логирование операций и SQL-запросов для отладки.

Установка

pip install dbalchemycore

Требуемые зависимости:

• sqlalchemy[asyncpg]
• pydantic
• pydantic-settings
• alembic

Структура проекта

project/
├── migration/                    # Директория для миграций Alembic
│   ├── versions/               # Файлы миграций
│   └── alembic.ini             # Конфигурация Alembic
├── dbAlchemyCore/              # Модуль библиотеки
│   ├── __init__.py             # Инициализация (init_db)
│   ├── models/                 # Модели SQLAlchemy
│   ├── schemas/                # Pydantic-схемы
│   ├── repositories/           # Репозитории для CRUD
│   ├── services/               # Бизнес-логика
│   └── utils/                  # Утилиты
├── .env                        # Переменные окружения
└── main.py                     # Точка входа

Начало работы

1. Настройка окружения: Создайте .env файл с параметрами подключения к базе данных:

DB_USER=postgres
DB_PASSWORD=secret
DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydb
DB_DRIVER=asyncpg
DB_DIALECT=postgresql
DB_POOL_SIZE=20
DB_MAX_OVERFLOW=50
DB_ECHO=False
DB_CONNECT_TIMEOUT=10

2. Инициализация базы данных: Вызовите init_db один раз при старте приложения, флаг use_create_all=True означает, что таблицы будут автоматически созданы через Metadata:

import asyncio
from dbAlchemyCore import init_db

async def main():
    await init_db(use_create_all=True)

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

3. Создание модели: Модели должны наследоваться от Base и использовать Mapped для полей:

# dbAlchemyCore/models/user.py
from sqlalchemy.orm import Mapped, mapped_column
from dbAlchemyCore import Base

class User(Base):
    name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[str] = mapped_column(unique=True, nullable=False)

Аннотации для часто используемых полей

id_field (автоинкрементный первичный ключ), created_at (дата создания) и updated_at (дата обновления). Они минимизируют дублирование кода и упрощают создание моделей. Пример использования

from sqlalchemy.orm import Mapped
from dbalchemycore.models import Base, id_field, created_at, updated_at

class User(Base):
    id: Mapped[id_field]
    name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[str] = mapped_column(unique=True, nullable=False)
    created_at: Mapped[created_at]
    updated_at: Mapped[updated_at]

Аннотации

from typing import Annotated
from sqlalchemy import Integer, DateTime, func
from sqlalchemy.orm import mapped_column

id_field = Annotated[int, mapped_column(Integer, primary_key=True, autoincrement=True)]
created_at = Annotated[DateTime, mapped_column(server_default=func.now())]
updated_at = Annotated[DateTime, mapped_column(server_default=func.now(), onupdate=func.now())]

4. Создание Pydantic-схем: Определите схемы для валидации данных:

# dbAlchemyCore/schemas/user_schemas.py
from pydantic import BaseModel
from typing import Optional

class UserCreate(BaseModel):
    name: str
    email: str

class UserFilter(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None

5. Создание репозитория: Создайте класс репозитория, унаследованный от BaseRepository:

# dbAlchemyCore/repositories/user_repo.py
from dbAlchemyCore.models.user import User
from dbAlchemyCore.repositories import BaseRepository

class UserRepo(BaseRepository):
    model = User

6. Использование: Пример выполнения CRUD-операций:

# main.py
import asyncio
from dbAlchemyCore import init_db
from dbAlchemyCore.repositories.user_repo import UserRepo
from dbAlchemyCore.schemas.user_schemas import UserCreate, UserFilter
from sqlalchemy.ext.asyncio import AsyncSession

async def main():
    await init_db(use_create_all=True)

    # Создание пользователя
    user_data = UserCreate(name="John", email="[e-mail]")
    await UserRepo.create(values=user_data, session=session)

    # Получение пользователя
    user = await UserRepo.get_one(filters=UserFilter(name="John"), session=session)
    print(user)  # {"id": 1, "name": "John", "email": "[e-mail]"}

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

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

init_db

Инициализирует подключение к базе данных, создаёт движок AsyncEngine и фабрику сессий async_sessionmaker. Cоздаёт таблицы при наличие флага use_create_all.

Параметры:

• use_create_all: bool — True, если необходимо, чтобы таблицы создавались автоматически через Metadata.create_all().

Исключения:

• ConnectionRefusedError: Сервер базы данных недоступен.
• InvalidPasswordError: Неверный пароль.
• InvalidCatalogNameError: Несуществующая база данных.
• OperationalError, DatabaseError, InvalidRequestError: Ошибки подключения SQLAlchemy.
• CommandError, FileNotFoundError: Ошибки миграций Alembic.

Base

Базовый класс для моделей, генерирует имя таблицы (например, users для класса User).

BaseRepository

Обобщённый класс для CRUD-операций. Поддерживает:

• Методы: get_one, get_many, create, update, delete, execute_sql.
• Фильтры: Через Pydantic-схемы, поддержка IN для списков (например, role=["admin", "moderator"]).
• Сортировка: Поддержка order_by с префиксом - для DESC.
• Пагинация: Параметры limit и offset.
• Группировка и агрегация: group_by, aggregations (count, sum, avg, min, max), having_filters.

Исключения:

• NotFoundError: Запись не найдена.
• EmptyFilterError: Пустые фильтры.
• InvalidFieldError: Несуществующее поле.
• EmptyValueError: Пустые значения для обновления.
• UnknowAggregationFunc: Неверная функция агрегации.
• MultipleResultsFound: Найдено несколько записей при strict=True в get_one.

connection()

Декоратор для управления асинхронными сессиями в кастомных методах. Поддерживает настройку уровня изоляции (isolation_level) и коммита (commit).

Пример:

from dbAlchemyCore.repositories import BaseRepository, connection
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepo(BaseRepository):
    model = User

    @classmethod
    @connection(commit=True)
    async def get_user_orm(cls, id: int, session: AsyncSession):
        return await session.get(cls.model, id)

Рекомендации

• Модели: Наследуйте от Base, используйте Mapped для полей.
• Репозитории: Создавайте отдельный класс для каждой модели, указывайте model.
• Pydantic-схемы: Используйте для валидации фильтров и значений, делайте поля опциональными для фильтрации/обновления.
• Транзакции: Применяйте @connection() для кастомных методов с доступом к базе.
• Бизнес-логика: Выносите сложную логику в модуль services.
• Тестирование: Используйте pytest-asyncio для асинхронных тестов, настройте тестовую БД (например, SQLite в памяти).

Примечания

• Асинхронность: Все операции асинхронные, используют AsyncSession и asyncpg.
• Ограничения: HAVING требует GROUP BY (ограничение SQL). Фильтры не поддерживают сравнения (>, <, и т.д.).
• Производительность: Используйте SQLAlchemy Core для bulk-операций, избегайте смешивания с ORM для предотвращения stale data.