Database lib Async

Обертка над SqlAlchemy для успешного асинхронного взаимодействия с базой данных

Параметры

Методы класса

Атрибуты класса

Example

import logging
import asyncio

from fastapi import FastAPI, Depends
from sqlalchemy import Integer, String, MetaData
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column, declarative_base

from async_sqlalchemy_bridge import AsyncDataBase, EngineProps, CrudRepository

Base = declarative_base(metadata=MetaData(schema="public"))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - [%(levelname)s] - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s",
    datefmt='%H:%M:%S',
)

logger = logging.getLogger(__name__)


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment="Идентификатор записи")
    first_name: Mapped[str] = mapped_column(String(30), comment="Имя", nullable=False)
    last_name: Mapped[str] = mapped_column(String(50), comment="Фамилия", nullable=False)


class Repo(CrudRepository):
    _table = User


def init_db() -> AsyncDataBase:
    props = EngineProps.initialize("postgresql+asyncpg://test:test@localhost:5432/test", "example_project")
    db = AsyncDataBase(props)
    db.init_echo_compiled_query(logger)
    return db


async def create_tables(db: AsyncDataBase) -> None:
    async with db.engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all, checkfirst=True)


async def main():
    db: AsyncDataBase = init_db()
    await create_tables(db)
    async with db.get_session() as session:
        repo = Repo(session)
        user = User(
            first_name="Ivan",
            last_name="Ivanov"
        )
        await repo.create(user)
        await session.commit()


def some_fastapi():
    app = FastAPI()
    db: AsyncDataBase = init_db()

    @app.put("/add")
    async def root(first_name: str, last_name: str, session: AsyncSession = Depends(db.get_dep_session)) -> int:
        repo = Repo(session)
        user = User(
            first_name=first_name,
            last_name=last_name
        )
        await repo.create(user)
        await session.commit()
        return user.id


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