refactor and using protocol

Author: Andrei Neagu

packages/service-library/src/servicelib/fastapi/lifespan_utils.py

@@ -1,13 +1,16 @@
-from collections.abc import AsyncIterator, Callable
-from typing import TypeAlias
+from collections.abc import AsyncIterator
+from typing import Protocol
 
 from fastapi import FastAPI
 from fastapi_lifespan_manager import LifespanManager, State
 
-SetupGenerator: TypeAlias = Callable[[FastAPI], AsyncIterator[State]]
 
+class LifespanGenerator(Protocol):
+    def __call__(self, app: FastAPI) -> AsyncIterator["State"]:
+        ...
 
-def combine_lifespans(*generators: SetupGenerator) -> LifespanManager:
+
+def combine_lifespans(*generators: LifespanGenerator) -> LifespanManager:
 
     manager = LifespanManager()
 

packages/service-library/src/servicelib/fastapi/tracing.py

@@ -17,7 +17,7 @@
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
-from servicelib.fastapi.lifespan_utils import SetupGenerator
+from servicelib.fastapi.lifespan_utils import LifespanGenerator
 from servicelib.logging_utils import log_context
 from settings_library.tracing import TracingSettings
 from yarl import URL
@@ -136,7 +136,7 @@ def setup_tracing(
 
 def get_lifespan_tracing(
     tracing_settings: TracingSettings, service_name: str
-) -> SetupGenerator:
+) -> LifespanGenerator:
     async def _(app: FastAPI) -> AsyncIterator[State]:
         setup_tracing(app, tracing_settings, service_name)
         yield {}

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/application.py

@@ -2,7 +2,7 @@
 
 from fastapi import FastAPI
 from fastapi_lifespan_manager import State
-from servicelib.fastapi.lifespan_utils import SetupGenerator, combine_lifespans
+from servicelib.fastapi.lifespan_utils import LifespanGenerator, combine_lifespans
 from servicelib.fastapi.openapi import override_fastapi_openapi_method
 from servicelib.fastapi.profiler import initialize_profiler
 from servicelib.fastapi.prometheus_instrumentation import (
@@ -43,7 +43,7 @@ async def _lifespan_banner(_: FastAPI) -> AsyncIterator[State]:
 def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
     app_settings = settings or ApplicationSettings.create_from_envs()
 
-    lifespans: list[SetupGenerator] = [
+    lifespans: list[LifespanGenerator] = [
         lifespan_director_v2,
         lifespan_director_v0,
         lifespan_rabbitmq,

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/notifier/_setup.py

@@ -1,7 +1,7 @@
-from servicelib.fastapi.lifespan_utils import SetupGenerator
+from servicelib.fastapi.lifespan_utils import LifespanGenerator
 
 from . import _notifier, _socketio
 
 
-def get_notifier_lifespans() -> list[SetupGenerator]:
+def get_notifier_lifespans() -> list[LifespanGenerator]:
     return [_socketio.lifespan, _notifier.lifespan]