jsdio/voice_assistant
Template
1
0
Fork 0
mirror of https://github.com/ijaric/voice_assistant.git synced 2025-05-24 14:33:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: provide db settings

Browse Source
This commit is contained in:
Artem Litvinov 2023-09-23 21:38:41 +01:00
parent 8bc6972ca2
commit bee0048a61
2 changed files with 13 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/fastapi_app/lib/app/settings.py
View File

@ -8,7 +8,7 @@ import lib.app.split_settings as app_split_settings
class Settings(pydantic_settings.BaseSettings):
api: app_split_settings.ApiSettings = pydantic.Field(default_factory=lambda: app_split_settings.ApiSettings())
db: app_split_settings.PostgresSettings = pydantic.Field(
postgres: app_split_settings.PostgresSettings = pydantic.Field(
default_factory=lambda: app_split_settings.PostgresSettings()
)
logger: app_split_settings.LoggingSettings = pydantic.Field(

22
src/fastapi_app/lib/db/postgres.py
View File

@ -2,30 +2,32 @@ import typing
import sqlalchemy.ext.asyncio as sa_asyncio
import lib.app.settings as app_settings
import lib.app.split_settings as app_split_settings
class AsyncDB:
"""Async DB connection."""
def __init__(self, settings: app_settings.Settings):
def __init__(self, settings: app_split_settings.DBSettings):
self.engine = sa_asyncio.create_async_engine(
url=settings.db.dsn,
pool_size=settings.db.pool_size,
pool_pre_ping=settings.db.pool_pre_ping,
echo=settings.db.echo,
url=settings.dsn,
pool_size=settings.pool_size,
pool_pre_ping=settings.pool_pre_ping,
echo=settings.echo,
future=True,
)
self.async_session = sa_asyncio.async_sessionmaker(
bind=self.engine,
autocommit=settings.db.auto_commit,
autoflush=settings.db.auto_flush,
expire_on_commit=settings.db.expire_on_commit,
autocommit=settings.auto_commit,
autoflush=settings.auto_flush,
expire_on_commit=settings.expire_on_commit,
class_=sa_asyncio.AsyncSession,
)
async def get_session(settings: app_settings.Settings) -> typing.AsyncGenerator[sa_asyncio.AsyncSession, typing.Any]:
async def get_session(
settings: app_split_settings.DBSettings,
) -> typing.AsyncGenerator[sa_asyncio.AsyncSession, typing.Any]:
db = AsyncDB(settings)
async with db.async_session() as session:
try: