aria2_server.db.server_config

__all__ module-attribute

__all__ = ('ServerConfigDatabase', 'get_server_config_db')

ServerConfigDatabase

ServerConfigDatabase(
    session: AsyncSession,
    server_config_table: Type[ServerConfig],
)

Source code in src/aria2_server/db/server_config/__init__.py

def __init__(self, session: AsyncSession, server_config_table: Type[ServerConfig]):
    self.session = session
    self.server_config_table = server_config_table

session instance-attribute

session = session

server_config_table instance-attribute

server_config_table = server_config_table

get async

get() -> ServerConfig

Source code in src/aria2_server/db/server_config/__init__.py

async def get(self) -> ServerConfig:
    results = await self.session.execute(select(self.server_config_table))
    server_config = results.scalar_one_or_none()  # only allow one row
    if server_config is None:
        return await self._create()
    return server_config

update async

update(update_dict: Dict[str, Any]) -> ServerConfig

Source code in src/aria2_server/db/server_config/__init__.py

async def update(self, update_dict: Dict[str, Any]) -> ServerConfig:
    server_config = await self.get()
    for key, value in update_dict.items():
        setattr(server_config, key, value)
    await self.session.commit()
    await self.session.refresh(server_config)
    return server_config

get_server_config_db async

get_server_config_db(
    session: AsyncSession = Depends(get_async_session),
) -> AsyncGenerator[ServerConfigDatabase, None]

Source code in src/aria2_server/db/server_config/__init__.py

async def get_server_config_db(
    session: AsyncSession = Depends(get_async_session),
) -> AsyncGenerator[ServerConfigDatabase, None]:
    yield ServerConfigDatabase(session, ServerConfig)