feat: Cosmos DB Change Feed Mode Support

[swapnil-nagar]

Summary

Updated Python Azure Functions library to support Cosmos DB change feed modes (LatestVersion and AllVersionsAndDeletes), aligning with Azure WebJobs Extensions upstream implementation.

Changes

• Type Definitions: Added CosmosDBChangeFeedMode type hint in _cosmosdb.py supporting 'LatestVersion' | 'AllVersionsAndDeletes'
• Module Documentation: Updated CosmosDBConverter and CosmosDBTriggerConverter class docstrings to document change feed mode behavior
• Public API: Exported CosmosDBChangeFeedMode type from main __init__.py for public consumption
• Tests: Added 3 new unit tests validating change feed mode support:
  • test_cosmosdb_change_feed_mode_latest_version_support: Validates LatestVersion mode
  • test_cosmosdb_change_feed_mode_all_versions_and_deletes_support: Validates AllVersionsAndDeletes mode
  • test_cosmosdb_change_feed_mode_type_available: Verifies type availability in public API

Files Modified

• azure/functions/_cosmosdb.py: Added CosmosDBChangeFeedMode type
• azure/functions/cosmosdb.py: Updated documentation, exported type
• azure/functions/__init__.py: Exported and documented CosmosDBChangeFeedMode
• tests/test_cosmosdb.py: Added change feed mode tests

Alignment

This change mirrors the Node.js library v4.16.1 Cosmos DB updates, ensuring consistent change feed support across both runtimes.

[swapnil-nagar]

/azp run

[azure-pipelines]

Commenter does not have sufficient privileges for PR 346 in repo Azure/azure-functions-python-library

[jviau]

I am not familiar with E2E-test situation for this repo, but want to make sure the feature is validated either manually or by tests. A specific scenario to validate for is the incoming payload type when using AllVersionsAndDelete. With LatestVersion it will be T, for AVAD it will be ChangeFeedItem<T>. So the JSON representation of this CSharp class: https://learn.microsoft.com/en-us/dotnet/api/microsoft.azure.cosmos.changefeeditem-1?view=azure-dotnet-preview

[hallvictoria]

For Jacob's comment, our Cosmos DB E2E tests are defined here, and we use the Cosmos DB emulator.

If this feature is supported in the emulator, you can create a PR to the python worker repo with the new function apps & tests. If the feature is not supported, we can create a follow-up task to update the Cosmos tests when the emulators have this change. Please let me know if you have any questions for testing!

[hallvictoria]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[jviau]

@hallvictoria I don't think it is supported in the emulator 😞. That has been the challenge with all my E2E tests also. I manually test with a real CosmosDB then disable the tests before checking in.

[swapnil-nagar]

I am not familiar with E2E-test situation for this repo, but want to make sure the feature is validated either manually or by tests. A specific scenario to validate for is the incoming payload type when using AllVersionsAndDelete. With LatestVersion it will be T, for AVAD it will be ChangeFeedItem<T>. So the JSON representation of this CSharp class: https://learn.microsoft.com/en-us/dotnet/api/microsoft.azure.cosmos.changefeeditem-1?view=azure-dotnet-preview

I validate the changes after getting access to feature. Here is the sample code.

Once these changes are merged and library is release we will create a sample repo for CosmosDb team, but below is the sample which is working

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

import json
import logging

import azure.functions as func

app = func.FunctionApp()


# Latest-version change feed (default mode).
# Equivalent to the Node.js "cosmosLatestVersion" sample.
@app.function_name(name="cosmosLatestVersion")
@app.cosmos_db_trigger(
    arg_name="documents",
    connection="CosmosConnection",
    database_name="%DATABASE_NAME%",
    container_name="%COSMOS_CONTAINER_NAME%",
    lease_container_name="%LEASE_CONTAINER_NAME%",
    lease_container_prefix="py-latest-version",
    create_lease_container_if_not_exists=True,
    start_from_beginning=True,
    change_feed_mode=func.CosmosDBChangeFeedMode.LATEST_VERSION,
)
def cosmos_latest_version(documents: func.DocumentList) -> None:
    logging.info("LatestVersion batch count=%d", len(documents))
    for document in documents:
        item = dict(document)
        logging.info(
            "LatestVersion item id=%s status=%s",
            item.get("id"),
            item.get("status"),
        )


# All-versions-and-deletes change feed (full-fidelity mode).
# Equivalent to the Node.js "cosmosFullFidelity" sample.
# NOTE: Unlike the Node.js CosmosDBChangeFeedItem (which exposes
# current/previous/metadata.operationType), the Python DocumentList binding
# delivers each change as a flat document (current image) that additionally
# carries the change-feed "_lsn" (log sequence number) field.
@app.function_name(name="cosmosFullFidelity")
@app.cosmos_db_trigger(
    arg_name="changes",
    connection="CosmosConnection",
    database_name="%DATABASE_NAME%",
    container_name="%COSMOS_CONTAINER_NAME%",
    lease_container_name="%LEASE_CONTAINER_NAME%",
    lease_container_prefix="py-full-fidelity",
    create_lease_container_if_not_exists=True,
    change_feed_mode=func.CosmosDBChangeFeedMode.ALL_VERSIONS_AND_DELETES,
)
def cosmos_full_fidelity(changes: func.DocumentList) -> None:
    logging.info("FullFidelity batch count=%d", len(changes))
    for index, change in enumerate(changes):
        item = dict(change)
        logging.info(
            "FullFidelity change index=%d id=%s status=%s lsn=%s",
            index,
            item.get("id"),
            item.get("status"),
            item.get("_lsn"),
        )
        logging.info("FullFidelity raw[%d]=%s", index, json.dumps(item, default=str))

[swapnil-nagar]

/azp run

[azure-pipelines]

Commenter does not have sufficient privileges for PR 346 in repo Azure/azure-functions-python-library

[hallvictoria]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[hallvictoria]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).