refactor: make resource pool initialisation async (#490)

Author: Panaetius

bases/renku_data_services/background_jobs/core.py

@@ -26,6 +26,11 @@
 from renku_data_services.namespace.models import NamespaceKind
 
 
+async def generate_user_namespaces(config: SyncConfig) -> None:
+    """Generate namespaces for users if there are none."""
+    await config.group_repo.generate_user_namespaces()
+
+
 async def sync_user_namespaces(config: SyncConfig) -> None:
     """Lists all user namespaces in the database and adds them to Authzed and the event queue."""
     authz = Authz(config.authz_config)

bases/renku_data_services/background_jobs/main.py

@@ -10,6 +10,7 @@
 from renku_data_services.background_jobs.core import (
     bootstrap_user_namespaces,
     fix_mismatched_project_namespace_ids,
+    generate_user_namespaces,
     migrate_groups_make_all_public,
     migrate_storages_v2_to_data_connectors,
     migrate_user_namespaces_make_all_public,
@@ -27,6 +28,7 @@ async def short_period_sync() -> None:
 
     await error_handler(
         [
+            generate_user_namespaces(config),
             bootstrap_user_namespaces(config),
             config.syncer.events_sync(config.kc_api),
             sync_admins_from_keycloak(config.kc_api, Authz(config.authz_config)),

bases/renku_data_services/data_api/main.py

@@ -69,7 +69,7 @@ def create_app() -> Sanic:
         # NOTE: in single process mode where we usually run schemathesis to get coverage the db migrations
         # specified below with the main_process_start decorator do not run.
         run_migrations_for_app("common")
-        config.rp_repo.initialize(config.db.conn_url(async_client=False), config.default_resource_pool)
+        asyncio.run(config.rp_repo.initialize(config.db.conn_url(async_client=False), config.default_resource_pool))
         asyncio.run(config.kc_user_repo.initialize(config.kc_api))
         asyncio.run(sync_admins_from_keycloak(config.kc_api, config.authz))
     if config.sentry.enabled:
@@ -122,10 +122,7 @@ async def setup_sentry(_: Sanic) -> None:
     async def do_migrations(_: Sanic) -> None:
         logger.info("running migrations")
         run_migrations_for_app("common")
-        config.rp_repo.initialize(config.db.conn_url(async_client=False), config.default_resource_pool)
-        await config.kc_user_repo.initialize(config.kc_api)
-        await sync_admins_from_keycloak(config.kc_api, config.authz)
-        await config.group_repo.generate_user_namespaces()
+        await config.rp_repo.initialize(config.db.conn_url(async_client=False), config.default_resource_pool)
 
     @app.before_server_start
     async def setup_rclone_validator(app: Sanic) -> None:

components/renku_data_services/crc/db.py

@@ -12,9 +12,9 @@
 from functools import wraps
 from typing import Any, Concatenate, Optional, ParamSpec, TypeVar, cast
 
-from sqlalchemy import NullPool, create_engine, delete, select
-from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import Session, selectinload, sessionmaker
+from sqlalchemy import NullPool, delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.orm import selectinload
 from sqlalchemy.sql import Select, and_, not_, or_
 from sqlalchemy.sql.expression import false, true
 
@@ -147,17 +147,16 @@ async def decorated_function(self: Any, *args: _P.args, **kwargs: _P.kwargs) ->
 class ResourcePoolRepository(_Base):
     """The adapter used for accessing resource pools with SQLAlchemy."""
 
-    def initialize(self, sync_connection_url: str, rp: models.ResourcePool) -> None:
+    async def initialize(self, async_connection_url: str, rp: models.ResourcePool) -> None:
         """Add the default resource pool if it does not already exist."""
-        engine = create_engine(sync_connection_url, poolclass=NullPool)
-        session_maker = sessionmaker(
+        engine = create_async_engine(async_connection_url, poolclass=NullPool)
+        session_maker = async_sessionmaker(
             engine,
-            class_=Session,
             expire_on_commit=True,
         )
-        with session_maker() as session, session.begin():
+        async with session_maker() as session, session.begin():
             stmt = select(schemas.ResourcePoolORM.default == true())
-            res = session.execute(stmt)
+            res = await session.execute(stmt)
             default_rp = res.scalars().first()
             if default_rp is None:
                 orm = schemas.ResourcePoolORM.load(rp)