use generics

Author: giancarloromeo

packages/celery-library/tests/conftest.py

@@ -73,7 +73,7 @@ def celery_settings(
 
 @pytest.fixture
 def app_server() -> BaseAppServer:
-    return FakeAppServer()
+    return FakeAppServer(app=None)
 
 
 @pytest.fixture(scope="session")

packages/service-library/src/servicelib/fastapi/queued_tasks/app_server.py

@@ -10,21 +10,15 @@
 _STARTUP_TIMEOUT: Final[float] = timedelta(minutes=1).total_seconds()
 
 
-class FastAPIAppServer(BaseAppServer):
+class FastAPIAppServer(BaseAppServer[FastAPI]):
     def __init__(self, app: FastAPI):
-        super().__init__()
-        self._app = app
+        super().__init__(app)
         self._lifespan_manager = LifespanManager(
-            self.fastapi_app,
+            app,
             startup_timeout=_STARTUP_TIMEOUT,
             shutdown_timeout=_SHUTDOWN_TIMEOUT,
         )
 
-    @property
-    def fastapi_app(self) -> FastAPI:
-        assert isinstance(self._app, FastAPI)  # nosec
-        return self._app
-
     async def on_startup(self) -> None:
         await self._lifespan_manager.__aenter__()
 

packages/service-library/src/servicelib/queued_tasks/app_server.py

@@ -3,32 +3,23 @@
 import threading
 from abc import ABC, abstractmethod
 from asyncio import AbstractEventLoop
-from contextlib import suppress
-from typing import TYPE_CHECKING, Final
+from typing import Final, Generic, TypeVar
 
 from servicelib.queued_tasks.task_manager import TaskManager
 
-if TYPE_CHECKING:
-    with suppress(ImportError):
-        from fastapi import FastAPI
-    with suppress(ImportError):
-        from aiohttp.web import Application
-
-
 STARTUP_TIMEOUT: Final[float] = datetime.timedelta(minutes=1).total_seconds()
 
+AppType = TypeVar("AppType")
 
-class BaseAppServer(ABC):
-    def __init__(self) -> None:
-        self._shutdown_event: asyncio.Event | None = None
 
-    @property
-    def fastapi_app(self) -> "FastAPI":
-        raise NotImplementedError
+class BaseAppServer(ABC, Generic[AppType]):
+    def __init__(self, app: AppType) -> None:
+        self._app: AppType = app
+        self._shutdown_event: asyncio.Event | None = None
 
     @property
-    def aiohttp_app(self) -> "Application":
-        raise NotImplementedError
+    def app(self) -> AppType:
+        return self._app
 
     @property
     def event_loop(self) -> AbstractEventLoop:

services/storage/src/simcore_service_storage/api/_worker_tasks/_files.py

@@ -1,6 +1,6 @@
 import logging
 
-from celery import Task  # type: ignore[import-untyped]
+from celery import Task
 from celery_library.utils import get_app_server
 from models_library.api_schemas_storage.storage_schemas import (
     FileUploadCompletionBody,
@@ -30,7 +30,7 @@ async def complete_upload_file(
         logging.INFO,
         msg=f"completing upload of file {user_id=}, {location_id=}, {file_id=}",
     ):
-        dsm = get_dsm_provider(get_app_server(task.app).fastapi_app).get(location_id)
+        dsm = get_dsm_provider(get_app_server(task.app).app).get(location_id)
         # NOTE: completing a multipart upload on AWS can take up to several minutes
         # if it returns slow we return a 202 - Accepted, the client will have to check later
         # for completeness

services/storage/src/simcore_service_storage/api/_worker_tasks/_paths.py

@@ -25,7 +25,7 @@ async def compute_path_size(
         logging.INFO,
         msg=f"computing path size {user_id=}, {location_id=}, {path=}",
     ):
-        dsm = get_dsm_provider(get_app_server(task.app).fastapi_app).get(location_id)
+        dsm = get_dsm_provider(get_app_server(task.app).app).get(location_id)
         return await dsm.compute_path_size(user_id, path=Path(path))
 
 
@@ -42,7 +42,7 @@ async def delete_paths(
         logging.INFO,
         msg=f"delete {paths=} in {location_id=} for {user_id=}",
     ):
-        dsm = get_dsm_provider(get_app_server(task.app).fastapi_app).get(location_id)
+        dsm = get_dsm_provider(get_app_server(task.app).app).get(location_id)
         files_ids: set[StorageFileID] = {
             TypeAdapter(StorageFileID).validate_python(f"{path}") for path in paths
         }

services/storage/src/simcore_service_storage/api/_worker_tasks/_simcore_s3.py

@@ -40,7 +40,7 @@ async def deep_copy_files_from_project(
         logging.INFO,
         msg=f"copying {body.source['uuid']} -> {body.destination['uuid']} with {task.request.id}",
     ):
-        dsm = get_dsm_provider(get_app_server(task.app).fastapi_app).get(
+        dsm = get_dsm_provider(get_app_server(task.app).app).get(
             SimcoreS3DataManager.get_location_id()
         )
         assert isinstance(dsm, SimcoreS3DataManager)  # nosec
@@ -75,7 +75,7 @@ async def export_data(
         logging.INFO,
         f"'{task_id}' export data (for {user_id=}) fom selection: {paths_to_export}",
     ):
-        dsm = get_dsm_provider(get_app_server(task.app).fastapi_app).get(
+        dsm = get_dsm_provider(get_app_server(task.app).app).get(
             SimcoreS3DataManager.get_location_id()
         )
         assert isinstance(dsm, SimcoreS3DataManager)  # nosec