Provide the ability to pass a session as a function in the repository.

[Kavan72]

Currently, there is no way to pass a session object at runtime. Ideally, we should be able to create a repository object and pass a session or inject a session while calling a repository method. We have 45 services and 57 repositories, and this feature could help us access any service from any other service and do the same with repositories.

Before I became familiar with this library, I was implementing it like this:

class Database:
    session: AsyncSession

    def __init__(
        self,
        database: str,
        username: str,
        password: str,
        host: str,
        port: int = 5432,
        echo: bool = True,
    ) -> None:
        self.url = URL.create(
            drivername="postgresql+asyncpg",
            username=username,
            password=password,
            host=host,
            port=port,
            database=database,
        ).render_as_string(hide_password=False)
        self.engine = create_async_engine(
            self.url,
            echo=echo,
            echo_pool="debug",
        )
        self.session_maker: async_sessionmaker[AsyncSession] = async_sessionmaker(
            self.engine, expire_on_commit=False
        )

class DatabaseRepository:
    """Repository for performing database queries."""

    def __init__(self, app, model: type[Model]):
        self.app = app
        self.model = model

    async def execute(self, *args, **kwargs):
        return await self.app.database.session.execute(*args, **kwargs)
        
class UserService:
    _model = User

    def __init__(self, app):
        self.app = app
        self.repository = DatabaseRepository(app, self._model)
        
async def get_db_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
    async with request.app.database.session_maker() as session:
        try:
            request.app.database.session = session
            yield session
            await session.commit()
        except exc.SQLAlchemyError as error:
            logger.debug(f"Rolling back due to exception {error}.")
            await session.rollback()
            raise error
        finally:
            await session.close()
            request.app.database.session = None

@router.delete("/user/{userId}")
async def delete_user_handler(
    request: Request,
    session: AsyncSession = Depends(get_db_session),
    user_id: int = Path(..., alias="userId"),
    token: AccessToken = Depends(JWTAPIAccessTokenBearer()),
):
    if not await request.app.services.user_service.delete_user(user_id):
        raise NotFound("User not found")
    return Response(status_code=204)

Loading the Database State

The database is loaded like this:

app.database = Database(
    app.config.POSTGRES_DB,
    app.config.POSTGRES_USER,
    app.config.POSTGRES_PASSWORD,
    app.config.POSTGRES_HOST,
    echo=app.config.ENVIRONMENT != ENVIRONMENT.PRODUCTION,
)

Loading Services into the Application State

Services are loaded into the application state during startup using a for loop.

[Priestch]

I'm also struggling to use other services in one service, some business logic involved multiple database models and using relationship is not enough.

Currently, I'm trying to abstract a universal layer to place all the business logic there, it has methods to provide all the advanced-alchemy services needed or use the services injected in the router handler. The layer will be initialized with database session, it combines the services to do the business. I'm not sure whether it follows the best practice.

If some business logic involved with multiple database models, what's the best practice?