sqlalchemy-example/database/context_async.py at main · ViktorViskov/sqlalchemy-example · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncEngine


# DB_CONNECTION_STRING = "sqlite+aiosqlite:///example.db"
# DB_CONNECTION_STRING = "postgresql+asyncpg://user:password@localhost:5432/dbname"
# DB_CONNECTION_STRING = "mariadb+asyncmy://user:password@localhost:3306/dbname"


class DbContextAsync:
    def __init__(self, connection_string: str):
        self._engine: AsyncEngine = create_async_engine(
            connection_string,
            echo=False,
        )

        self._session_maker = async_sessionmaker(
            bind=self._engine, expire_on_commit=False
        )

        # PostgreSQL / MySQL engine
        # self._engine = create_engine(
        #     connection_string, echo=False, pool_pre_ping=True, pool_recycle=3600
        # )

    @property
    def session(self) -> async_sessionmaker[AsyncSession]:
        return self._session_maker

    @property
    def engine(self) -> AsyncEngine:
        return self._engine

    async def close(self) -> None:
        await self._engine.dispose()