40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
from typing import AsyncGenerator
|
|
|
|
from litestar.contrib.sqlalchemy.plugins import EngineConfig, SQLAlchemyAsyncConfig
|
|
from litestar.exceptions import ClientException
|
|
from litestar.status_codes import HTTP_409_CONFLICT
|
|
from sqlalchemy import URL
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
|
|
async def provide_transaction(
|
|
db_session: AsyncSession,
|
|
) -> AsyncGenerator[AsyncSession, None]:
|
|
try:
|
|
async with db_session.begin():
|
|
yield db_session
|
|
except IntegrityError as exc:
|
|
raise ClientException(
|
|
status_code=HTTP_409_CONFLICT,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
|
|
sessionmaker = async_sessionmaker(expire_on_commit=False)
|
|
|
|
db_connection_string = URL.create(
|
|
drivername="postgresql+asyncpg",
|
|
username="televend",
|
|
password="televend",
|
|
host="localhost",
|
|
port=5433,
|
|
database="televend",
|
|
)
|
|
db_config = SQLAlchemyAsyncConfig(
|
|
connection_string=db_connection_string.render_as_string(hide_password=False),
|
|
engine_config=EngineConfig(
|
|
echo=True,
|
|
),
|
|
)
|