{"id":10595,"date":"2025-06-06T08:00:26","date_gmt":"2025-06-06T15:00:26","guid":{"rendered":"https:\/\/devblogs.microsoft.com\/cosmosdb\/?p=10595"},"modified":"2025-06-04T07:41:37","modified_gmt":"2025-06-04T14:41:37","slug":"building-a-modern-python-api-with-azure-cosmos-db-a-5-part-video-series","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/cosmosdb\/building-a-modern-python-api-with-azure-cosmos-db-a-5-part-video-series\/","title":{"rendered":"Building a Modern Python API with Azure Cosmos DB: A 5-Part Video Series"},"content":{"rendered":"<div>\n<div>I&#8217;m excited to share our new video series where I walk through building a production-ready inventory management API using Python, FastAPI, and Azure Cosmos DB NoSQL. This project demonstrates modern async patterns, clean architecture, and enterprise-grade features like batch operations and optimistic concurrency control.<\/div>\n<div><\/div>\n<div>This builds on our <a href=\"https:\/\/devblogs.microsoft.com\/cosmosdb\/getting-started-with-azure-cosmos-db-using-the-python-sdk\/\">Getting Started Series<\/a>, check out those videos to grasp the fundamentals first.<\/div>\n<div><\/div>\n<h3>Video 1: Data Modeling with Pydantic<\/h3>\n<\/div>\n<div>\n<h2><iframe title=\"YouTube video player\" src=\"\/\/www.youtube.com\/embed\/9sJGxigUido?si=7c9i77JoLA91AwOm\" width=\"560\" height=\"315\" frameborder=\"0\" allowfullscreen=\"allowfullscreen\"><\/iframe><\/h2>\n<p>In the first video, I dive deep into data modeling using Pydantic v2, showing how to create robust, type-safe models with built-in validation. The key insight is using model inheritance to separate concerns between what clients send and what the database returns.<\/p>\n<div>\n<div>\n<pre class=\"prettyprint language-py\"><code class=\"language-py\">class Product(BaseModel):\r\n    \"\"\"Fields that are intrinsic to a product\"\"\"\r\n    name: Annotated[str, Field(min_length=1, max_length=255)]\r\n    description: Annotated[Optional[str], Field(None, max_length=1000)]\r\n    category: Annotated[str, Field(min_length=1, max_length=100)] \r\n    price: Annotated[Decimal, Field(gt=0, decimal_places=2)]\r\n    sku: Annotated[str, Field(min_length=1, max_length=50, pattern=r\"^[A-Z0-9-]+$\")]\r\n    quantity: Annotated[int, Field(ge=0)] = 0\r\n    status: ProductStatus = ProductStatus.ACTIVE\r\n\r\n    model_config = ConfigDict(extra=\"forbid\", str_strip_whitespace=True)\r\n\r\nclass ProductResponse(Product):\r\n    \"\"\"All product fields plus system-generated fields.\"\"\"\r\n    id: str\r\n    etag: str = Field(alias=\"_etag\")  # Cosmos DB concurrency control token\r\n    last_updated: datetime\r\n<\/code><\/pre>\n<div>\n<div>The model hierarchy ensures that input validation happens automatically, while response models include system-generated fields like ETags for optimistic concurrency control.<\/div>\n<div><\/div>\n<h3>Video 2: Client Configuration and Connection Management<\/h3>\n<h3><iframe src=\"\/\/www.youtube.com\/embed\/D7GKPIx_ARg\" width=\"560\" height=\"314\" allowfullscreen=\"allowfullscreen\"><\/iframe><\/h3>\n<p><span style=\"font-size: 16px;\">The second video covers setting up a production-ready Azure Cosmos DB client with proper authentication, connection pooling, and singleton pattern implementation. I show how to use Azure&#8217;s managed identity for secure, credential-free authentication.<\/span><\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n<div>\n<div>\n<pre class=\"prettyprint language-py\"><code class=\"language-py\">async def _ensure_client() -&gt; CosmosClient:\r\n    \"\"\"\r\n    Ensures a single CosmosClient instance exists for the lifetime of the application.\r\n    \"\"\"\r\n    global _client, _credential\r\n    if _client is None:\r\n        try:\r\n            logger.info(\"Initializing Cosmos DB client with DefaultAzureCredential\")\r\n            _credential = DefaultAzureCredential()\r\n\r\n            client_options = {\r\n                \"connection_timeout\": 60, \r\n            }\r\n\r\n            _client = CosmosClient(COSMOSDB_ENDPOINT, _credential, **client_options)\r\n\r\n            logger.info(\r\n                \"Cosmos DB client initialized successfully\",\r\n                extra={\r\n                    \"endpoint\": COSMOSDB_ENDPOINT,\r\n                    \"auth_method\": \"DefaultAzureCredential\",\r\n                },\r\n            )\r\n        except Exception as e:\r\n            logger.error(f\"Failed to initialize Cosmos DB client: {e}\")\r\n            raise\r\n    return _client<\/code><\/pre>\n<div>\n<div>This pattern ensures efficient resource usage while maintaining thread safety in async environments.<\/div>\n<\/div>\n<\/div>\n<h3>Video 3: Async Operations Done Right<\/h3>\n<h3><iframe src=\"\/\/www.youtube.com\/embed\/jmK5msNic9c\" width=\"560\" height=\"314\" allowfullscreen=\"allowfullscreen\"><\/iframe><\/h3>\n<p><span style=\"font-size: 16px;\">In the third video, I demonstrate how to leverage FastAPI&#8217;s async capabilities with Azure Cosmos DB&#8217;s async SDK. The combination of async\/await patterns with dependency injection creates clean, testable code.<\/span><\/p>\n<div>\n<pre class=\"prettyprint language-py\"><code class=\"language-py\">async def update_product(\r\n    container: ContainerProxy,\r\n    identifier: VersionedProductIdentifier,\r\n    updates: ProductUpdate\r\n) -&gt; ProductResponse:\r\n    \"\"\"Update an existing product with optimistic concurrency control.\"\"\"\r\n    update_dict = updates.model_dump(exclude_unset=True)\r\n    normalized_category = normalize_category(identifier.category)\r\n    update_dict[\"last_updated\"] = datetime.now(timezone.utc).isoformat()\r\n    \r\n    # Create patch operations for partial updates\r\n    patch_operations = []\r\n    for key, value in update_dict.items():\r\n        if key not in [\"id\", \"category\", \"_etag\"]: \r\n            patch_operations.append(\r\n                {\"op\": \"set\", \"path\": f\"\/{key}\", \"value\": value}\r\n            )\r\n    \r\n    try:\r\n        result = await container.patch_item(\r\n            item=identifier.id,\r\n            partition_key=normalized_category,\r\n            patch_operations=patch_operations,\r\n            headers={\"if-match\": identifier.etag},  # ETag for concurrency control\r\n        )\r\n        return ProductResponse.model_validate(result)\r\n    except Exception as e:\r\n        handle_cosmos_error(e, operation=\"update\", product_id=identifier.id)<\/code><\/pre>\n<\/div>\n<div>\n<div>\n<div>The use of patch operations enables efficient partial updates while ETags prevent concurrent modification conflicts.<\/div>\n<div><\/div>\n<h3>Video 4: Batch Operations for Performance<\/h3>\n<h3><iframe src=\"\/\/www.youtube.com\/embed\/I0zH-Zb8d7M\" width=\"560\" height=\"314\" allowfullscreen=\"allowfullscreen\"><\/iframe><\/h3>\n<p><span style=\"font-size: 16px;\">The fourth video explores batch operations, showing how to process multiple items efficiently using Azure Cosmos DB&#8217;s batch API and Python&#8217;s asyncio for concurrent execution across partitions.<\/span><\/p>\n<\/div>\n<\/div>\n<div>\n<pre class=\"prettyprint language-py\"><code class=\"language-py\">async def _execute_batch_by_category(\r\n    items_by_category: dict[str, list[T]],\r\n    process_func: Callable[[str, list[T]], Any],\r\n    result_extractor: Callable[[Any], list[R]] | None = None\r\n) -&gt; list[R] | list[Any]:\r\n    \"\"\"\r\n    Generic function to execute batch operations grouped by category \r\n    with concurrent processing.\r\n    \"\"\"\r\n    # Schedule tasks to run concurrently for each category\r\n    tasks = [\r\n        asyncio.create_task(process_func(category, items))\r\n        for category, items in items_by_category.items()\r\n    ]\r\n    \r\n    # Wait for all tasks to complete and gather results\r\n    results = await asyncio.gather(*tasks, return_exceptions=True)\r\n    \r\n    # Check for exceptions and collect successful results\r\n    all_results = []\r\n    exceptions = []\r\n    \r\n    for result in results:\r\n        if isinstance(result, Exception):\r\n            exceptions.append(result)\r\n        else:\r\n            if result_extractor and result is not None:\r\n                all_results.extend(result_extractor(result))\r\n    \r\n    if exceptions:\r\n        raise exceptions[0]\r\n    \r\n    return all_results<\/code><\/pre>\n<\/div>\n<div>\n<div>This generic approach allows efficient batch processing while respecting Azure Cosmos DB&#8217;s partition key requirements.<\/div>\n<div><\/div>\n<h3>Video 5: Centralized Error Handling and Logging<\/h3>\n<h3><iframe src=\"\/\/www.youtube.com\/embed\/e9K7_HPY2Aw\" width=\"560\" height=\"314\" allowfullscreen=\"allowfullscreen\"><\/iframe><\/h3>\n<p><span style=\"font-size: 16px;\">The final video demonstrates how to implement centralized error handling that transforms low-level database errors into meaningful application exceptions, complete with structured logging for observability.<\/span><\/p>\n<\/div>\n<div>\n<pre class=\"prettyprint language-py\"><code class=\"language-py\">def handle_cosmos_error(e: Exception, operation: str, **context: Any) -&gt; None:\r\n    \"\"\"Convert Cosmos DB exceptions to application-specific exceptions.\"\"\"\r\n    \r\n    if isinstance(e, CosmosBatchOperationError):\r\n        status_code = getattr(e, 'status_code', None)\r\n        if status_code == 409:\r\n            error_message = str(e).lower()\r\n            if \"unique index constraint\" in error_message and \"\/sku\" in error_message:\r\n                sku = context.get(\"sku\", \"unknown\")\r\n                raise ProductDuplicateSKUError(\r\n                    f\"Product with SKU '{sku}' already exists\"\r\n                ) from e\r\n            else:\r\n                # Handle ID conflicts - prefer showing SKU if available\r\n                sku = context.get(\"sku\")\r\n                if sku:\r\n                    raise ProductAlreadyExistsError(\r\n                        f\"Product with SKU '{sku}' already exists\"\r\n                    ) from e\r\n\r\n@app.exception_handler(ApplicationError)\r\nasync def application_error_handler(request: Request, exc: ApplicationError) -&gt; JSONResponse:\r\n    \"\"\"Handle application-specific errors with structured logging.\"\"\"\r\n    logger = logging.getLogger(request.scope.get(\"route\").endpoint.__module__)\r\n    \r\n    # Log at appropriate level based on status code\r\n    log_level = \"warning\" if exc.status_code &lt; 500 else \"error\"\r\n    extra = {\"error_type\": type(exc).__name__}\r\n    \r\n    getattr(logger, log_level)(str(exc), extra=extra, exc_info=exc)<\/code><\/pre>\n<div>\n<div>This approach ensures that errors are handled consistently across the application while providing meaningful feedback to API consumers.<\/div>\n<\/div>\n<\/div>\n<h2>Conclusion<\/h2>\n<div>This project demonstrates how modern Python features combined with Azure Cosmos DB can create a robust, scalable API. Key takeaways include:<\/div>\n<ul>\n<li>Leveraging Pydantic for strong typing and validation<\/li>\n<li>Implementing proper async patterns for optimal performance<\/li>\n<li>Using batch operations to minimize round trips to the database<\/li>\n<li>Building comprehensive error handling for production reliability<\/li>\n<li>Maintaining clean architecture with clear separation of concerns<\/li>\n<\/ul>\n<div>The <a href=\"https:\/\/github.com\/madebygps\/cosmosdb-fastapi-functions\" target=\"_blank\" rel=\"noopener\">complete source code is available on GitHub<\/a>, and I encourage you to watch the video series for detailed explanations. Whether you&#8217;re building a new API or modernizing an existing one, these patterns will help you create more maintainable and performant applications.\u00a0 Happy coding!<\/div>\n<h2><strong>Leave a review<\/strong><\/h2>\n<\/div>\n<div>\n<p>Tell us about your Azure Cosmos DB experience! Leave a review on PeerSpot and we\u2019ll gift you $50.\u00a0<a id=\"menuros8\" class=\"fui-Link ___1q1shib f2hkw1w f3rmtva f1ewtqcl fyind8e f1k6fduh f1w7gpdv fk6fouc fjoy568 figsok6 f1s184ao f1mk8lai fnbmjn9 f1o700av f13mvf36 f1cmlufx f9n3di6 f1ids18y f1tx3yz7 f1deo86v f1eh06m1 f1iescvh fhgqx19 f1olyrje f1p93eir f1nev41a f1h8hb77 f1lqvz6u f10aw75t fsle3fq f17ae5zn\" title=\"https:\/\/peerspotdotcom.my.site.com\/proreviews\/?salesopportunityproduct=00kpy000004tkxjia4&amp;productpeerspotnumber=30881&amp;calendlyaccount=peerspot&amp;calendlyformlink=peerspot-product-reviews-ps-gc-vi-sf-50&amp;giftcard=50\" href=\"https:\/\/peerspotdotcom.my.site.com\/proReviews\/?SalesOpportunityProduct=00kPy000004TKXJIA4&amp;productPeerspotNumber=30881&amp;CalendlyAccount=peerspot&amp;CalendlyFormLink=peerspot-product-reviews-ps-gc-vi-sf-50&amp;giftCard=50\" target=\"_blank\" rel=\"noreferrer noopener\" aria-label=\"Link Get started here\">Get started here<\/a>.<\/p>\n<h2 id=\"about-azure-cosmos-db\"><strong>About Azure Cosmos DB<\/strong><\/h2>\n<p>Azure Cosmos DB is a fully managed and serverless NoSQL and vector database for modern app development, including AI applications. With its SLA-backed speed and availability as well as instant dynamic scalability, it is ideal for real-time NoSQL and MongoDB applications that require high performance and distributed computing over massive volumes of NoSQL and vector data.<\/p>\n<p>To stay in the loop on Azure Cosmos DB updates, follow us on\u00a0<a id=\"menurosb\" class=\"fui-Link ___1q1shib f2hkw1w f3rmtva f1ewtqcl fyind8e f1k6fduh f1w7gpdv fk6fouc fjoy568 figsok6 f1s184ao f1mk8lai fnbmjn9 f1o700av f13mvf36 f1cmlufx f9n3di6 f1ids18y f1tx3yz7 f1deo86v f1eh06m1 f1iescvh fhgqx19 f1olyrje f1p93eir f1nev41a f1h8hb77 f1lqvz6u f10aw75t fsle3fq f17ae5zn\" title=\"https:\/\/twitter.com\/azurecosmosdb\" href=\"https:\/\/twitter.com\/AzureCosmosDB\" target=\"_blank\" rel=\"noreferrer noopener\" aria-label=\"Link X\">X<\/a>,\u00a0<a id=\"menurose\" class=\"fui-Link ___1q1shib f2hkw1w f3rmtva f1ewtqcl fyind8e f1k6fduh f1w7gpdv fk6fouc fjoy568 figsok6 f1s184ao f1mk8lai fnbmjn9 f1o700av f13mvf36 f1cmlufx f9n3di6 f1ids18y f1tx3yz7 f1deo86v f1eh06m1 f1iescvh fhgqx19 f1olyrje f1p93eir f1nev41a f1h8hb77 f1lqvz6u f10aw75t fsle3fq f17ae5zn\" title=\"https:\/\/aka.ms\/azurecosmosdbyoutube\" href=\"https:\/\/aka.ms\/AzureCosmosDBYouTube\" target=\"_blank\" rel=\"noreferrer noopener\" aria-label=\"Link YouTube\">YouTube<\/a>, and\u00a0<a id=\"menurosh\" class=\"fui-Link ___1q1shib f2hkw1w f3rmtva f1ewtqcl fyind8e f1k6fduh f1w7gpdv fk6fouc fjoy568 figsok6 f1s184ao f1mk8lai fnbmjn9 f1o700av f13mvf36 f1cmlufx f9n3di6 f1ids18y f1tx3yz7 f1deo86v f1eh06m1 f1iescvh fhgqx19 f1olyrje f1p93eir f1nev41a f1h8hb77 f1lqvz6u f10aw75t fsle3fq f17ae5zn\" title=\"https:\/\/www.linkedin.com\/company\/azure-cosmos-db\/\" href=\"https:\/\/www.linkedin.com\/company\/azure-cosmos-db\/\" target=\"_blank\" rel=\"noreferrer noopener\" aria-label=\"Link LinkedIn\">LinkedIn<\/a>.<\/p>\n<p>To quickly build your first database, watch our\u00a0<a href=\"https:\/\/youtube.com\/playlist?list=PLmamF3YkHLoLLGUtSoxmUkORcWaTyHlXp\" target=\"_blank\" rel=\"noopener\">Get Started videos<\/a>\u00a0on YouTube and explore ways to\u00a0<a href=\"https:\/\/docs.microsoft.com\/azure\/cosmos-db\/optimize-dev-test\" target=\"_blank\" rel=\"noopener\">dev\/test free.<\/a><\/p>\n<\/div>\n","protected":false},"excerpt":{"rendered":"<p>I&#8217;m excited to share our new video series where I walk through building a production-ready inventory management API using Python, FastAPI, and Azure Cosmos DB NoSQL. This project demonstrates modern async patterns, clean architecture, and enterprise-grade features like batch operations and optimistic concurrency control. This builds on our Getting Started Series, check out those videos [&hellip;]<\/p>\n","protected":false},"author":154186,"featured_media":10597,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[14,1217],"tags":[1793,1312],"class_list":["post-10595","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-core-sql-api","category-python-sdk","tag-azure-cosmos-db-python-sdk","tag-python"],"acf":[],"blog_post_summary":"<p>I&#8217;m excited to share our new video series where I walk through building a production-ready inventory management API using Python, FastAPI, and Azure Cosmos DB NoSQL. This project demonstrates modern async patterns, clean architecture, and enterprise-grade features like batch operations and optimistic concurrency control. This builds on our Getting Started Series, check out those videos [&hellip;]<\/p>\n","_links":{"self":[{"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/posts\/10595","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/users\/154186"}],"replies":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/comments?post=10595"}],"version-history":[{"count":0,"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/posts\/10595\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/media\/10597"}],"wp:attachment":[{"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/media?parent=10595"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/categories?post=10595"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/cosmosdb\/wp-json\/wp\/v2\/tags?post=10595"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}