Implement publish product workflow

Author: AndreuCodina

api/src/api/.env.Local

@@ -1 +1,4 @@
-LoggingLevel="INFO"
+LoggingLevel="INFO"
+CosmosDbNoSqlUrl="http://localhost:8081"
+CosmosDbNoSqlKey="C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw=="
+CosmosDbNoSqlDatabase="mydatabase"

api/src/api/application_settings.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 from azure.identity import DefaultAzureCredential
+from pydantic import SecretStr
 from pydantic.alias_generators import to_pascal
 from pydantic_settings import (
     AzureKeyVaultSettingsSource,
@@ -18,6 +19,9 @@ class ApplicationSettings(BaseSettings):
     model_config = SettingsConfigDict(alias_generator=to_pascal, extra="ignore")
 
     logging_level: str
+    cosmos_db_no_sql_url: str
+    cosmos_db_no_sql_key: SecretStr
+    cosmos_db_no_sql_database: str
 
     @classmethod
     def settings_customise_sources(

api/src/api/dependency_container.py

@@ -1,22 +1,48 @@
+from azure.cosmos import CosmosClient, PartitionKey
+
 from api.application_settings import ApplicationSettings
 from api.workflows.products.publish_product.publish_product_workflow import (
     PublishProductWorkflow,
 )
+from common.application_environment import ApplicationEnvironment
+from domain.entities.product import Product
 
 
 class DependencyContainer:
     @classmethod
     def initialize(cls) -> None:
         cls.initialize_application_settings()
+        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
+
+        with CosmosClient(
+            cls.get_application_settings().cosmos_db_no_sql_url,
+            cls.get_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
+            )
+            cosmosdb_database = cosmosdb_client.get_database_client(
+                cls.get_application_settings().cosmos_db_no_sql_database
+            )
+            cosmosdb_database.create_container_if_not_exists(
+                id=Product.__name__, partition_key=PartitionKey("/id")
+            )
+
     @classmethod
     def get_application_settings(cls) -> ApplicationSettings:
         return cls.application_settings
 
     @classmethod
     def get_publish_product_workflow(cls) -> PublishProductWorkflow:
-        return PublishProductWorkflow()
+        return PublishProductWorkflow(
+            application_settings=cls.get_application_settings()
+        )

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

@@ -1,11 +1,32 @@
+from azure.cosmos.aio import CosmosClient
+
+from api.application_settings import ApplicationSettings
 from api.workflows.products.publish_product.publish_product_request import (
     PublishProductRequest,
 )
 from api.workflows.products.publish_product.publish_product_response import (
     PublishProductResponse,
 )
+from domain.entities.product import Product
 
 
 class PublishProductWorkflow:
+    def __init__(self, application_settings: ApplicationSettings) -> None:
+        self.application_settings = application_settings
+
     async def execute(self, request: PublishProductRequest) -> PublishProductResponse:
-        return PublishProductResponse(id=request.name)
+        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)

domain/src/domain/entities/product.py

@@ -13,6 +13,12 @@ class Product(Entity):
     is_discontinued: bool
     discontinuation_reason: str | None = None
 
+    @staticmethod
+    def publish(name: str, price: Decimal, description: str | None = None) -> "Product":
+        return Product(
+            name=name, price=price, description=description, is_discontinued=False
+        )
+
     def discontinue(self, discontinuation_reason: str | None = None) -> None:
         if self.is_discontinued:
             error_message = "The product is already discontinued"

pyproject.toml

@@ -3,6 +3,8 @@ name = "python-architecture"
 version = "0.1.0"
 requires-python = ">=3.12"
 dependencies = [
+    "aiohttp==3.11.18",
+    "azure-cosmos==4.9.0",
     "common",
     "domain",
     "fastapi[standard]==0.115.12",