Make DependencyContainer async

Author: AndreuCodina

api/src/api/dependency_container.py

@@ -1,4 +1,8 @@
-from azure.cosmos import CosmosClient, PartitionKey
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+from azure.cosmos import PartitionKey
+from azure.cosmos.aio import CosmosClient, DatabaseProxy
 
 from api.application_settings import ApplicationSettings
 from api.workflows.products.discontinue_product.discontinue_product_workflow import (
@@ -13,30 +17,54 @@
 
 class DependencyContainer:
     @classmethod
-    def initialize(cls) -> None:
+    async def initialize(cls) -> None:
         cls.initialize_application_settings()
-        cls.initialize_local_environment()
+        await cls.initialize_local_environment()
 
     @classmethod
     def initialize_application_settings(cls) -> None:
         cls.application_settings = ApplicationSettings()  # type: ignore[reportCallIssue]
 
     @classmethod
-    def initialize_local_environment(cls) -> None:
-        if ApplicationEnvironment.get_current() != ApplicationEnvironment.LOCAL:
-            return
+    @asynccontextmanager
+    async def get_cosmosdb_client(
+        cls,
+    ) -> AsyncGenerator[CosmosClient]:
+        application_settings = cls.get_application_settings()
 
-        with CosmosClient(
-            cls.get_application_settings().cosmos_db_no_sql_url,
-            cls.get_application_settings().cosmos_db_no_sql_key.get_secret_value(),
+        async with CosmosClient(
+            application_settings.cosmos_db_no_sql_url,
+            application_settings.cosmos_db_no_sql_key.get_secret_value(),
         ) as cosmosdb_client:
-            cosmosdb_client.create_database_if_not_exists(
-                id=cls.get_application_settings().cosmos_db_no_sql_database
-            )
+            yield cosmosdb_client
+
+    @classmethod
+    @asynccontextmanager
+    async def get_cosmosdb_database(
+        cls,
+    ) -> AsyncGenerator[DatabaseProxy]:
+        application_settings = cls.get_application_settings()
+
+        async with cls.get_cosmosdb_client() as cosmosdb_client:
             cosmosdb_database = cosmosdb_client.get_database_client(
-                cls.get_application_settings().cosmos_db_no_sql_database
+                application_settings.cosmos_db_no_sql_database
             )
-            cosmosdb_database.create_container_if_not_exists(
+            yield cosmosdb_database
+
+    @classmethod
+    async def initialize_local_environment(cls) -> None:
+        if ApplicationEnvironment.get_current() != ApplicationEnvironment.LOCAL:
+            return
+
+        application_settings = cls.get_application_settings()
+
+        async with cls.get_cosmosdb_client() as cosmosdb_client:
+            await cosmosdb_client.create_database_if_not_exists(
+                application_settings.cosmos_db_no_sql_database
+            )
+
+        async with cls.get_cosmosdb_database() as cosmosdb_database:
+            await cosmosdb_database.create_container_if_not_exists(
                 id=Product.__name__, partition_key=PartitionKey("/id")
             )
 
@@ -45,13 +73,17 @@ def get_application_settings(cls) -> ApplicationSettings:
         return cls.application_settings
 
     @classmethod
-    def get_publish_product_workflow(cls) -> PublishProductWorkflow:
-        return PublishProductWorkflow(
-            application_settings=cls.get_application_settings()
-        )
+    @asynccontextmanager
+    async def get_publish_product_workflow(
+        cls,
+    ) -> AsyncGenerator[PublishProductWorkflow]:
+        async with cls.get_cosmosdb_database() as cosmosdb_database:
+            yield PublishProductWorkflow(cosmosdb_database=cosmosdb_database)
 
     @classmethod
-    def get_discontinue_product_workflow(cls) -> DiscontinueProductWorkflow:
-        return DiscontinueProductWorkflow(
-            application_settings=cls.get_application_settings()
-        )
+    @asynccontextmanager
+    async def get_discontinue_product_workflow(
+        cls,
+    ) -> AsyncGenerator[DiscontinueProductWorkflow]:
+        async with cls.get_cosmosdb_database() as cosmosdb_database:
+            yield DiscontinueProductWorkflow(cosmosdb_database=cosmosdb_database)

api/src/api/main.py

@@ -1,3 +1,6 @@
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from opentelemetry.instrumentation.fastapi import (  # type: ignore[reportMissingTypeStubs]
@@ -9,17 +12,27 @@
 from common.application_environment import ApplicationEnvironment
 
 
+@asynccontextmanager
+async def lifespan(_: FastAPI) -> AsyncGenerator[None]:
+    await DependencyContainer.initialize()
+    yield
+
+
 def add_telemetry(app: FastAPI) -> None:
     FastAPIInstrumentor.instrument_app(app)  # type: ignore[reportUnknownMemberType]
 
 
-DependencyContainer.initialize()
 openapi_url = (
     "/openapi.json"
     if ApplicationEnvironment.get_current() != ApplicationEnvironment.PRODUCTION
     else None
 )
-app = FastAPI(title="Python monorepo", version="0.1.0", openapi_url=openapi_url)
+app = FastAPI(
+    title="Python monorepo",
+    version="0.1.0",
+    openapi_url=openapi_url,
+    lifespan=lifespan,
+)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],

api/src/api/workflows/products/discontinue_product/discontinue_product_workflow.py

@@ -1,28 +1,22 @@
-from azure.cosmos.aio import CosmosClient
+from azure.cosmos.aio import DatabaseProxy
 
-from api.application_settings import ApplicationSettings
 from api.workflows.products.discontinue_product.discontinue_product_request import (
     DiscontinueProductRequest,
 )
 from domain.entities.product import Product
 
 
 class DiscontinueProductWorkflow:
-    def __init__(self, application_settings: ApplicationSettings) -> None:
-        self.application_settings = application_settings
+    def __init__(self, cosmosdb_database: DatabaseProxy) -> None:
+        self.cosmosdb_database = cosmosdb_database
 
     async def execute(self, request: DiscontinueProductRequest) -> None:
-        async with CosmosClient(
-            self.application_settings.cosmos_db_no_sql_url,
-            self.application_settings.cosmos_db_no_sql_key.get_secret_value(),
-        ) as cosmosdb_client:
-            cosmosdb_database = cosmosdb_client.get_database_client(
-                self.application_settings.cosmos_db_no_sql_database
-            )
-            product_container = cosmosdb_database.get_container_client(Product.__name__)
-            product_dict = await product_container.read_item(
-                item=request.id, partition_key=request.id
-            )
-            product = Product.model_validate(product_dict)
-            product.discontinue(request.discontinuation_reason)
-            await product_container.replace_item(product.id, product.model_dump())
+        product_container = self.cosmosdb_database.get_container_client(
+            Product.__name__
+        )
+        product_dict = await product_container.read_item(
+            item=request.id, partition_key=request.id
+        )
+        product = Product.model_validate(product_dict)
+        product.discontinue(request.discontinuation_reason)
+        await product_container.replace_item(product.id, product.model_dump())

api/src/api/workflows/products/product_router.py

@@ -19,9 +19,11 @@
 
 @router.post("")
 async def publish_product(request: PublishProductRequest) -> PublishProductResponse:
-    return await DependencyContainer.get_publish_product_workflow().execute(request)
+    async with DependencyContainer.get_publish_product_workflow() as workflow:
+        return await workflow.execute(request)
 
 
 @router.post("/discontinue")
 async def discontinue_product(request: DiscontinueProductRequest) -> None:
-    return await DependencyContainer.get_discontinue_product_workflow().execute(request)
+    async with DependencyContainer.get_discontinue_product_workflow() as workflow:
+        return await workflow.execute(request)

api/src/api/workflows/products/publish_product/publish_product_workflow.py

@@ -1,6 +1,5 @@
-from azure.cosmos.aio import CosmosClient
+from azure.cosmos.aio import DatabaseProxy
 
-from api.application_settings import ApplicationSettings
 from api.workflows.products.publish_product.publish_product_request import (
     PublishProductRequest,
 )
@@ -11,22 +10,17 @@
 
 
 class PublishProductWorkflow:
-    def __init__(self, application_settings: ApplicationSettings) -> None:
-        self.application_settings = application_settings
+    def __init__(self, cosmosdb_database: DatabaseProxy) -> None:
+        self.cosmosdb_database = cosmosdb_database
 
     async def execute(self, request: PublishProductRequest) -> PublishProductResponse:
-        async with CosmosClient(
-            self.application_settings.cosmos_db_no_sql_url,
-            self.application_settings.cosmos_db_no_sql_key.get_secret_value(),
-        ) as cosmosdb_client:
-            cosmosdb_database = cosmosdb_client.get_database_client(
-                self.application_settings.cosmos_db_no_sql_database
-            )
-            product_container = cosmosdb_database.get_container_client(Product.__name__)
-            product = Product.publish(
-                name=request.name,
-                description=request.description,
-                price=request.price,
-            )
-            await product_container.create_item(product.model_dump())
-            return PublishProductResponse(id=product.id)
+        product_container = self.cosmosdb_database.get_container_client(
+            Product.__name__
+        )
+        product = Product.publish(
+            name=request.name,
+            description=request.description,
+            price=request.price,
+        )
+        await product_container.create_item(product.model_dump())
+        return PublishProductResponse(id=product.id)

api/tests/integration/conftest.py

@@ -1,25 +1,8 @@
-from collections.abc import AsyncGenerator
-
 import pytest
-from azure.cosmos.aio import CosmosClient, DatabaseProxy
 
 from api.dependency_container import DependencyContainer
 
 
 @pytest.fixture(autouse=True)
-def initialize_dependency_container() -> None:
-    DependencyContainer.initialize()
-
-
-@pytest.fixture
-async def cosmosdb_database() -> AsyncGenerator[DatabaseProxy, None]:
-    application_settings = DependencyContainer.get_application_settings()
-
-    async with CosmosClient(
-        application_settings.cosmos_db_no_sql_url,
-        application_settings.cosmos_db_no_sql_key.get_secret_value(),
-    ) as cosmosdb_client:
-        cosmosdb_database = cosmosdb_client.get_database_client(
-            application_settings.cosmos_db_no_sql_database
-        )
-        yield cosmosdb_database
+async def initialize_dependency_container() -> None:
+    await DependencyContainer.initialize()

api/tests/integration/workflows/products/discontinue_product/test_discontinue_product_workflow.py

@@ -1,5 +1,4 @@
 import pytest
-from azure.cosmos.aio import DatabaseProxy
 
 from api.dependency_container import DependencyContainer
 from api.workflows.products.discontinue_product.discontinue_product_request import (
@@ -11,12 +10,15 @@
 
 @pytest.mark.integration
 class TestDiscontinueProductWorkflow:
-    async def test_discontinue_product(self, cosmosdb_database: DatabaseProxy) -> None:
-        product = ProductBuilder().build()
-        await cosmosdb_database.get_container_client(Product.__name__).create_item(
-            product.model_dump()
-        )
-        request = DiscontinueProductRequest(id=product.id)
-        workflow = DependencyContainer.get_discontinue_product_workflow()
+    async def test_discontinue_product(self) -> None:
+        async with DependencyContainer.get_cosmosdb_database() as cosmosdb_database:
+            product = ProductBuilder().build()
+            await cosmosdb_database.get_container_client(Product.__name__).create_item(
+                product.model_dump()
+            )
+            request = DiscontinueProductRequest(id=product.id)
 
-        await workflow.execute(request)
+            async with (
+                DependencyContainer.get_discontinue_product_workflow() as workflow
+            ):
+                await workflow.execute(request)

api/tests/integration/workflows/products/publish_product/test_publish_product_workflow.py

@@ -11,6 +11,5 @@ class TestPublishProductWorkflow:
     async def test_publish_product(self) -> None:
         request = PublishProductRequestBuilder().build()
 
-        workflow = DependencyContainer.get_publish_product_workflow()
-
-        await workflow.execute(request)
+        async with DependencyContainer.get_publish_product_workflow() as workflow:
+            await workflow.execute(request)