Integrate fastapi-sqla library into ALBS (AlmaLinux/build-system#230)

- Replaced sync, async and pulp databases dependencies with fastapi_sqla dependencies
- Integrated fastapi-sqla sqla session management with open_session/open_async_session
- Removed manual session management with .begin(), .close(), .commit()
- Setting up fastapi-sqla for scripts and dramatiq tasks outside FastAPI app
- Pytest fixtures for fastapi-sqla integration

alws/app.py

@@ -3,6 +3,7 @@
 
 import sentry_sdk
 from fastapi import FastAPI
+from fastapi_sqla import setup as fastapi_sqla_setup
 from starlette.middleware.exceptions import ExceptionMiddleware
 
 from alws import routers
@@ -34,6 +35,7 @@
 
 app = FastAPI()
 app.add_middleware(ExceptionMiddleware, handlers=handlers)
+fastapi_sqla_setup(app)
 
 for module in ROUTERS:
     for router_type in (

alws/auth/dependencies.py

@@ -1,15 +1,18 @@
 from fastapi import Depends
+from fastapi_sqla import AsyncSessionDependency
 from fastapi_users.authentication.strategy import (
     AccessTokenDatabase,
     DatabaseStrategy,
     JWTStrategy,
 )
 from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
-from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
+from fastapi_users_db_sqlalchemy.access_token import (
+    SQLAlchemyAccessTokenDatabase,
+)
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from alws.config import settings
-from alws.dependencies import get_async_session
+from alws.dependencies import get_async_db_key
 from alws.models import User, UserAccessToken, UserOauthAccount
 
 __all__ = [
@@ -20,19 +23,26 @@
 ]
 
 
-async def get_user_db(session: AsyncSession = Depends(get_async_session)):
+async def get_user_db(
+    session: AsyncSession = Depends(
+        AsyncSessionDependency(key=get_async_db_key())
+    ),
+):
     yield SQLAlchemyUserDatabase(
-        session, User, oauth_account_table=UserOauthAccount)
+        session, User, oauth_account_table=UserOauthAccount
+    )
 
 
 async def get_access_token_db(
-    session: AsyncSession = Depends(get_async_session),
+    session: AsyncSession = Depends(
+        AsyncSessionDependency(key=get_async_db_key())
+    ),
 ):
     yield SQLAlchemyAccessTokenDatabase(session, UserAccessToken)
 
 
 def get_database_strategy(
-        access_token_db: AccessTokenDatabase = Depends(get_access_token_db)
+    access_token_db: AccessTokenDatabase = Depends(get_access_token_db),
 ) -> DatabaseStrategy:
     return DatabaseStrategy(access_token_db, lifetime_seconds=3600)
 

alws/config.py

@@ -26,16 +26,38 @@ class Settings(BaseSettings):
 
     redis_url: str = 'redis://redis:6379'
 
-    database_url: str = (
-        'postgresql+asyncpg://postgres:password@db/almalinux-bs'
-    )
+    # TBD: remove after moving to fastapi-sqla
+    database_url: str = 'postgresql+asyncpg://postgres:password@db/almalinux-bs'
     test_database_url: str = (
         'postgresql+asyncpg://postgres:password@db/test-almalinux-bs'
     )
+    # TBD: remove after moving to fastapi-sqla
     sync_database_url: str = (
         'postgresql+psycopg2://postgres:password@db/almalinux-bs'
     )
 
+    fastapi_sqla__async__sqlalchemy_url: str = (
+        'postgresql+asyncpg://postgres:password@db/almalinux-bs'
+    )
+    fastapi_sqla__async__sqlalchemy_echo_pool: bool = True
+
+    sqlalchemy_url: str = (
+        'postgresql+psycopg2://postgres:password@db/almalinux-bs'
+    )
+    sqlalchemy_pool_pre_ping: bool = True
+    sqlalchemy_pool_recycle: int = 3600
+
+    fastapi_sqla__pulp__sqlalchemy_url: str = (
+        'postgresql+psycopg2://postgres:password@pulp:5432/pulp'
+    )
+    fastapi_sqla__pulp__sqlalchemy_pool_pre_ping: bool = True
+    fastapi_sqla__pulp__sqlalchemy_pool_recycle: int = 3600
+
+    fastapi_sqla__pulp_async__sqlalchemy_url: str = (
+        'postgresql+asyncpg://postgres:password@pulp:5432/pulp'
+    )
+    fastapi_sqla__pulp_async__sqlalchemy_echo_pool: bool = True
+
     github_client: str
     github_client_secret: str
 

alws/crud/actions.py

@@ -1,11 +1,11 @@
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.future import select
 
-from alws.database import Session
 from alws.models import UserAction
 from alws.perms.actions import ActionsList
 
 
-async def ensure_all_actions_exist(session: Session, commit: bool = False):
+async def ensure_all_actions_exist(session: AsyncSession, commit: bool = False):
     existing_actions = (
         (await session.execute(select(UserAction))).scalars().all()
     )
@@ -19,7 +19,4 @@ async def ensure_all_actions_exist(session: Session, commit: bool = False):
 
     if new_actions:
         session.add_all(new_actions)
-        if commit:
-            await session.commit()
-        else:
-            await session.flush()
+        await session.flush()

alws/crud/build.py

@@ -85,7 +85,7 @@ async def create_build(
         for flavour in flavors:
             db_build.platform_flavors.append(flavour)
     db.add(db_build)
-    await db.commit()
+    await db.flush()
     await db.refresh(db_build)
     start_build.send(db_build.id, build.model_dump())
     return db_build
@@ -175,9 +175,7 @@ async def generate_query(count=False):
                 sqlalchemy.or_(
                     models.BuildTaskRef.url.like(f"%/{project_name}.git"),
                     models.BuildTaskRef.url.like(f"%/{project_name}%.src.rpm"),
-                    models.BuildTaskRef.url.like(
-                        f"%/rpms/{project_name}%.git"
-                    ),
+                    models.BuildTaskRef.url.like(f"%/rpms/{project_name}%.git"),
                 )
             )
             if not (await db.execute(project_query)).scalars().all():
@@ -201,13 +199,11 @@ async def generate_query(count=False):
         if build_task_arch is not None:
             query = query.filter(models.BuildTask.arch == build_task_arch)
         if any(rpm_params.values()):
-            pulp_params.update(
-                {
-                    key: value
-                    for key, value in rpm_params.items()
-                    if value is not None
-                }
-            )
+            pulp_params.update({
+                key: value
+                for key, value in rpm_params.items()
+                if value is not None
+            })
             # TODO: we can get packages from pulp database
             pulp_hrefs = await pulp_client.get_rpm_packages(**pulp_params)
             pulp_hrefs = [row["pulp_href"] for row in pulp_hrefs]
@@ -293,128 +289,113 @@ async def remove_build_job(db: AsyncSession, build_id: int):
     build_task_ref_ids = []
     test_task_ids = []
     test_task_artifact_ids = []
-    async with db.begin():
-        build = await db.execute(query_bj)
-        build = build.scalars().first()
-        if build is None:
-            raise DataNotFoundError(f'Build with {build_id} not found')
-        if build.products:
-            product_names = "\n".join(
-                (product.name for product in build.products)
-            )
-            raise BuildError(
-                f"Cannot delete Build={build_id}, "
-                f"build contains in following products:\n{product_names}"
-            )
-        if build.released:
-            raise BuildError(f"Build with {build_id} is released")
-        for bt in build.tasks:
-            build_task_ids.append(bt.id)
-            build_task_ref_ids.append(bt.ref_id)
-            for build_artifact in bt.artifacts:
-                build_task_artifact_ids.append(build_artifact.id)
-            for tt in bt.test_tasks:
-                test_task_ids.append(tt.id)
-                repo_ids.append(tt.repository_id)
-                for test_artifact in tt.artifacts:
-                    test_task_artifact_ids.append(test_artifact.id)
-        for br in build.repos:
-            repos.append(br.pulp_href)
-            repo_ids.append(br.id)
-        pulp_client = PulpClient(
-            settings.pulp_host, settings.pulp_user, settings.pulp_password
-        )
-        await db.execute(
-            delete(models.BuildRepo).where(
-                models.BuildRepo.c.build_id == build_id
-            )
-        )
-        await db.execute(
-            delete(models.BuildPlatformFlavour).where(
-                models.BuildPlatformFlavour.c.build_id == build_id
-            )
-        )
-        await db.execute(
-            delete(models.SignTask).where(models.SignTask.build_id == build_id)
-        )
-        await db.execute(
-            delete(models.BinaryRpm).where(
-                models.BinaryRpm.build_id == build_id
-            )
+    build = await db.execute(query_bj)
+    build = build.scalars().first()
+    if build is None:
+        raise DataNotFoundError(f'Build with {build_id} not found')
+    if build.products:
+        product_names = "\n".join((product.name for product in build.products))
+        raise BuildError(
+            f"Cannot delete Build={build_id}, "
+            f"build contains in following products:\n{product_names}"
         )
-        await db.execute(
-            delete(models.SourceRpm).where(
-                models.SourceRpm.build_id == build_id
-            )
-        )
-        await db.execute(
-            delete(models.PerformanceStats).where(
-                models.PerformanceStats.build_task_id.in_(build_task_ids)
-            )
-        )
-        await db.execute(
-            delete(models.PerformanceStats).where(
-                models.PerformanceStats.test_task_id.in_(test_task_ids)
-            )
-        )
-        await db.execute(
-            delete(models.TestTaskArtifact).where(
-                models.TestTaskArtifact.id.in_(test_task_artifact_ids)
-            )
-        )
-        await db.execute(
-            delete(models.TestTask).where(
-                models.TestTask.id.in_(test_task_ids)
-            )
+    if build.released:
+        raise BuildError(f"Build with {build_id} is released")
+    for bt in build.tasks:
+        build_task_ids.append(bt.id)
+        build_task_ref_ids.append(bt.ref_id)
+        for build_artifact in bt.artifacts:
+            build_task_artifact_ids.append(build_artifact.id)
+        for tt in bt.test_tasks:
+            test_task_ids.append(tt.id)
+            repo_ids.append(tt.repository_id)
+            for test_artifact in tt.artifacts:
+                test_task_artifact_ids.append(test_artifact.id)
+    for br in build.repos:
+        repos.append(br.pulp_href)
+        repo_ids.append(br.id)
+    pulp_client = PulpClient(
+        settings.pulp_host, settings.pulp_user, settings.pulp_password
+    )
+    await db.execute(
+        delete(models.BuildRepo).where(models.BuildRepo.c.build_id == build_id)
+    )
+    await db.execute(
+        delete(models.BuildPlatformFlavour).where(
+            models.BuildPlatformFlavour.c.build_id == build_id
         )
-        await db.execute(
-            delete(models.BuildTaskArtifact).where(
-                models.BuildTaskArtifact.id.in_(build_task_artifact_ids)
-            )
+    )
+    await db.execute(
+        delete(models.SignTask).where(models.SignTask.build_id == build_id)
+    )
+    await db.execute(
+        delete(models.BinaryRpm).where(models.BinaryRpm.build_id == build_id)
+    )
+    await db.execute(
+        delete(models.SourceRpm).where(models.SourceRpm.build_id == build_id)
+    )
+    await db.execute(
+        delete(models.PerformanceStats).where(
+            models.PerformanceStats.build_task_id.in_(build_task_ids)
         )
-        await db.execute(
-            delete(models.BuildTaskDependency).where(
-                models.BuildTaskDependency.c.build_task_dependency.in_(
-                    build_task_ids
-                )
-            )
+    )
+    await db.execute(
+        delete(models.PerformanceStats).where(
+            models.PerformanceStats.test_task_id.in_(test_task_ids)
         )
-        await db.execute(
-            delete(models.Repository).where(models.Repository.id.in_(repo_ids))
+    )
+    await db.execute(
+        delete(models.TestTaskArtifact).where(
+            models.TestTaskArtifact.id.in_(test_task_artifact_ids)
         )
-        await db.execute(
-            delete(models.BuildTask).where(
-                models.BuildTask.build_id == build_id
-            )
+    )
+    await db.execute(
+        delete(models.TestTask).where(models.TestTask.id.in_(test_task_ids))
+    )
+    await db.execute(
+        delete(models.BuildTaskArtifact).where(
+            models.BuildTaskArtifact.id.in_(build_task_artifact_ids)
         )
-        await db.execute(
-            delete(models.BuildDependency).where(
-                sqlalchemy.or_(
-                    models.BuildDependency.c.build_dependency == build_id,
-                    models.BuildDependency.c.build_id == build_id,
-                )
+    )
+    await db.execute(
+        delete(models.BuildTaskDependency).where(
+            models.BuildTaskDependency.c.build_task_dependency.in_(
+                build_task_ids
             )
         )
-        await db.execute(
-            delete(models.BuildTaskRef).where(
-                models.BuildTaskRef.id.in_(build_task_ref_ids)
+    )
+    await db.execute(
+        delete(models.Repository).where(models.Repository.id.in_(repo_ids))
+    )
+    await db.execute(
+        delete(models.BuildTask).where(models.BuildTask.build_id == build_id)
+    )
+    await db.execute(
+        delete(models.BuildDependency).where(
+            sqlalchemy.or_(
+                models.BuildDependency.c.build_dependency == build_id,
+                models.BuildDependency.c.build_id == build_id,
             )
         )
-        await db.execute(
-            delete(models.Build).where(models.Build.id == build_id)
+    )
+    await db.execute(
+        delete(models.BuildTaskRef).where(
+            models.BuildTaskRef.id.in_(build_task_ref_ids)
         )
-        # FIXME
-        # it seems we cannot just delete any files because
-        # https://docs.pulpproject.org/pulpcore/restapi.html#tag/Content:-Files
-        # does not content delete option, but artifact does:
-        # https://docs.pulpproject.org/pulpcore/restapi.html#operation/
-        # artifacts_delete
-        # "Remove Artifact only if it is not associated with any Content."
-        # for artifact in artifacts:
-        # await pulp_client.remove_artifact(artifact)
-        for repo in repos:
-            try:
-                await pulp_client.delete_by_href(repo, wait_for_result=True)
-            except Exception as err:
-                logging.exception("Cannot delete repo from pulp: %s", err)
-        await db.commit()
+    )
+    await db.execute(delete(models.Build).where(models.Build.id == build_id))
+    # FIXME
+    # it seems we cannot just delete any files because
+    # https://docs.pulpproject.org/pulpcore/restapi.html#tag/Content:-Files
+    # does not content delete option, but artifact does:
+    # https://docs.pulpproject.org/pulpcore/restapi.html#operation/
+    # artifacts_delete
+    # "Remove Artifact only if it is not associated with any Content."
+    # for artifact in artifacts:
+    # await pulp_client.remove_artifact(artifact)
+    for repo in repos:
+        try:
+            await pulp_client.delete_by_href(repo, wait_for_result=True)
+        except Exception as err:
+            logging.exception("Cannot delete repo from pulp: %s", err)
+    await db.flush()