diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index b8b1c451ad10..bedd74df141b 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -3,6 +3,7 @@ ### 4.16.2 (Unreleased) #### Features Added +* `query_items_change_feed()` now returns a `CosmosItemPaged` (sync) / `CosmosAsyncItemPaged` (async), matching `query_items()`. This exposes the thread-safe public `get_response_headers()` API, allowing users to read the change feed continuation token (`etag`) from the latest page's response headers instead of relying on the non-thread-safe internal `client_connection.last_response_headers`. See [PR 47947](https://github.com/Azure/azure-sdk-for-python/pull/47947). #### Breaking Changes diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py index a5d01a7a12c9..ff468c7f6488 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py @@ -1243,7 +1243,7 @@ def QueryItemsChangeFeed( options: Optional[Mapping[str, Any]] = None, response_hook: Optional[Callable[[Mapping[str, Any], Mapping[str, Any]], None]] = None, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Queries documents change feed in a collection. :param str collection_link: The link to the document collection. @@ -1252,9 +1252,11 @@ def QueryItemsChangeFeed( :type response_hook: Callable[[dict[str, str], dict[str, Any]] :return: - Query Iterable of Documents. + A pageable of documents from the change feed. In addition to iteration, it exposes + get_response_headers() for thread-safe access to the response headers of the most + recent page fetch (including the change feed continuation token, ``etag``). :rtype: - query_iterable.QueryIterable + ~azure.cosmos.CosmosItemPaged """ @@ -1272,7 +1274,7 @@ def _QueryChangeFeed( partition_key_range_id: Optional[str] = None, response_hook: Optional[Callable[[Mapping[str, Any], Mapping[str, Any]], None]] = None, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Queries change feed of a resource in a collection. :param str collection_link: The link to the document collection. @@ -1282,9 +1284,11 @@ def _QueryChangeFeed( :type response_hook: Callable[[dict[str, str], dict[str, Any]] :return: - Query Iterable of Documents. + A pageable of documents from the change feed. In addition to iteration, it exposes + get_response_headers() for thread-safe access to the response headers of the most + recent page fetch (including the change feed continuation token, ``etag``). :rtype: - query_iterable.QueryIterable + ~azure.cosmos.CosmosItemPaged """ if options is None: @@ -1295,6 +1299,9 @@ def _QueryChangeFeed( path = base.GetPathFromLink(collection_link, http_constants.ResourceType.Document) collection_id = base.GetResourceIdOrFullNameFromLink(collection_link) + # Shared dict for header capture — overwritten each page fetch + response_headers: CaseInsensitiveDict = CaseInsensitiveDict() + def fetch_fn(options: Mapping[str, Any]) -> Tuple[list[dict[str, Any]], CaseInsensitiveDict]: if collection_link in self.__container_properties_cache: # TODO: This will make deep copy. Check if this has any performance impact @@ -1311,14 +1318,16 @@ def fetch_fn(options: Mapping[str, Any]) -> Tuple[list[dict[str, Any]], CaseInse options, partition_key_range_id, response_hook=response_hook, + response_headers=response_headers, **kwargs) - return ItemPaged( + return CosmosItemPaged( self, options, fetch_function=fetch_fn, collection_link=collection_link, - page_iterator_class=ChangeFeedIterable + page_iterator_class=ChangeFeedIterable, + response_headers=response_headers ) def _ReadPartitionKeyRanges( diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py index b4d760b5c5f4..92f76a74970a 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py @@ -985,7 +985,7 @@ def query_items_change_feed( response_hook: Optional[Callable[[Mapping[str, str], dict[str, Any]], None]] = None, availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. :keyword int max_item_count: Max number of items to be returned in the enumeration operation. @@ -1020,8 +1020,8 @@ def query_items_change_feed( If not provided, uses the client's configured strategy. :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An AsyncItemPaged of items (dicts). - :rtype: AsyncItemPaged[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosAsyncItemPaged """ ... @@ -1037,7 +1037,7 @@ def query_items_change_feed( response_hook: Optional[Callable[[Mapping[str, Any], dict[str, Any]], None]] = None, availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. :keyword dict[str, Any] feed_range: The feed range that is used to define the scope. @@ -1068,8 +1068,8 @@ def query_items_change_feed( If not provided, uses the client's configured strategy. :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An AsyncItemPaged of items (dicts). - :rtype: AsyncItemPaged[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosAsyncItemPaged """ ... @@ -1083,7 +1083,7 @@ def query_items_change_feed( availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, response_hook: Optional[Callable[[Mapping[str, Any], dict[str, Any]], None]] = None, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. :keyword str continuation: The continuation token retrieved from previous response. It contains chang feed mode. @@ -1104,8 +1104,8 @@ def query_items_change_feed( If not provided, uses the client's configured strategy. :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An AsyncItemPaged of items (dicts). - :rtype: AsyncItemPaged[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosAsyncItemPaged """ # pylint: enable=line-too-long ... @@ -1121,7 +1121,7 @@ def query_items_change_feed( availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, response_hook: Optional[Callable[[Mapping[str, Any], dict[str, Any]], None]] = None, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Get a sorted list of items that were changed in the entire container, in the order in which they were modified. @@ -1152,8 +1152,8 @@ def query_items_change_feed( If not provided, uses the client's configured strategy. :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An AsyncItemPaged of items (dicts). - :rtype: AsyncItemPaged[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosAsyncItemPaged """ ... @@ -1161,7 +1161,7 @@ def query_items_change_feed( def query_items_change_feed( # pylint: disable=unused-argument self, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. @@ -1197,8 +1197,8 @@ def query_items_change_feed( # pylint: disable=unused-argument If not provided, uses the client's configured strategy. :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An AsyncItemPaged of items (dicts). - :rtype: AsyncItemPaged[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosAsyncItemPaged """ # pylint: disable=too-many-statements validate_kwargs(kwargs) diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py index 1f9f2ca87369..a7cbb927b53b 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py @@ -2459,7 +2459,7 @@ def QueryItemsChangeFeed( options: Optional[Mapping[str, Any]] = None, response_hook: Optional[Callable[[Mapping[str, Any], Mapping[str, Any]], None]] = None, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Queries documents change feed in a collection. :param str collection_link: The link to the document collection. @@ -2467,9 +2467,11 @@ def QueryItemsChangeFeed( :param response_hook: A callable invoked with the response metadata. :type response_hook: Callable[[dict[str, str], dict[str, Any]] :return: - Query Iterable of Documents. + A pageable of documents from the change feed. In addition to async iteration, it exposes + get_response_headers() for thread-safe access to the response headers of the most + recent page fetch (including the change feed continuation token, ``etag``). :rtype: - query_iterable.QueryIterable + ~azure.cosmos.aio.CosmosAsyncItemPaged """ @@ -2489,7 +2491,7 @@ def _QueryChangeFeed( partition_key_range_id: Optional[str] = None, response_hook: Optional[Callable[[Mapping[str, Any], Mapping[str, Any]], None]] = None, **kwargs: Any - ) -> AsyncItemPaged[dict[str, Any]]: + ) -> CosmosAsyncItemPaged: """Queries change feed of a resource in a collection. :param str collection_link: The link to the document collection. @@ -2499,9 +2501,11 @@ def _QueryChangeFeed( :param response_hook: A callable invoked with the response metadata :type response_hook: Callable[[dict[str, str], dict[str, Any]] :return: - Query Iterable of Documents. + A pageable of documents from the change feed. In addition to async iteration, it exposes + get_response_headers() for thread-safe access to the response headers of the most + recent page fetch (including the change feed continuation token, ``etag``). :rtype: - query_iterable.QueryIterable + ~azure.cosmos.aio.CosmosAsyncItemPaged """ if options is None: @@ -2519,6 +2523,9 @@ def _QueryChangeFeed( path = base.GetPathFromLink(collection_link, resource_key) collection_id = base.GetResourceIdOrFullNameFromLink(collection_link) + # Shared dict for header capture — overwritten each page fetch + response_headers: CaseInsensitiveDict = CaseInsensitiveDict() + async def fetch_fn(options: Mapping[str, Any]) -> Tuple[list[dict[str, Any]], CaseInsensitiveDict]: if collection_link in self.__container_properties_cache: new_options = dict(options) @@ -2535,17 +2542,19 @@ async def fetch_fn(options: Mapping[str, Any]) -> Tuple[list[dict[str, Any]], Ca options, partition_key_range_id, response_hook=response_hook, + response_headers=response_headers, **kwargs ), self.last_response_headers, ) - return AsyncItemPaged( + return CosmosAsyncItemPaged( self, options, fetch_function=fetch_fn, collection_link=collection_link, - page_iterator_class=ChangeFeedIterable + page_iterator_class=ChangeFeedIterable, + response_headers=response_headers ) def QueryOffers( diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py index 665ae2e4f869..fbba41e0c2ff 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py @@ -471,7 +471,7 @@ def query_items_change_feed( availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, response_hook: Optional[Callable[[Mapping[str, str], dict[str, Any]], None]] = None, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. :keyword int max_item_count: Max number of items to be returned in the enumeration operation. @@ -508,8 +508,8 @@ def query_items_change_feed( :paramtype availability_strategy: Union[bool, dict[str, Any]] :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An Iterable of items (dicts). - :rtype: Iterable[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosItemPaged """ ... @@ -525,7 +525,7 @@ def query_items_change_feed( availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, response_hook: Optional[Callable[[Mapping[str, str], dict[str, Any]], None]] = None, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. @@ -558,8 +558,8 @@ def query_items_change_feed( :paramtype availability_strategy: Union[bool, dict[str, Any]] :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An Iterable of items (dicts). - :rtype: Iterable[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosItemPaged """ ... @@ -573,7 +573,7 @@ def query_items_change_feed( availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, response_hook: Optional[Callable[[Mapping[str, str], dict[str, Any]], None]] = None, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. :keyword str continuation: The continuation token retrieved from previous response. It contains chang feed mode. @@ -595,8 +595,8 @@ def query_items_change_feed( :paramtype availability_strategy: Union[bool, dict[str, Any]] :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An Iterable of items (dicts). - :rtype: Iterable[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosItemPaged """ ... @@ -611,7 +611,7 @@ def query_items_change_feed( availability_strategy: Optional[Union[bool, dict[str, Any]]] = None, response_hook: Optional[Callable[[Mapping[str, str], dict[str, Any]], None]] = None, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Get a sorted list of items that were changed in the entire container, in the order in which they were modified, @@ -643,8 +643,8 @@ def query_items_change_feed( :paramtype availability_strategy: Union[bool, dict[str, Any]] :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] - :returns: An Iterable of items (dicts). - :rtype: Iterable[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosItemPaged """ ... @@ -653,7 +653,7 @@ def query_items_change_feed( self, *args: Any, **kwargs: Any - ) -> ItemPaged[dict[str, Any]]: + ) -> CosmosItemPaged: """Get a sorted list of items that were changed, in the order in which they were modified. :keyword str continuation: The continuation token retrieved from previous response. It contains chang feed mode. @@ -692,8 +692,8 @@ def query_items_change_feed( :keyword response_hook: A callable invoked with the response metadata. :paramtype response_hook: Callable[[Mapping[str, str], dict[str, Any]], None] :param Any args: args - :returns: An Iterable of items (dicts). - :rtype: Iterable[dict[str, Any]] + :returns: A pageable of items (dicts) exposing get_response_headers() for the latest page's headers. + :rtype: CosmosItemPaged """ # pylint: disable=too-many-statements diff --git a/sdk/cosmos/azure-cosmos/tests/test_query_response_headers.py b/sdk/cosmos/azure-cosmos/tests/test_query_response_headers.py index 73fe64342f09..5f6c2e2a64fe 100644 --- a/sdk/cosmos/azure-cosmos/tests/test_query_response_headers.py +++ b/sdk/cosmos/azure-cosmos/tests/test_query_response_headers.py @@ -957,6 +957,49 @@ def hook(headers, _body): finally: self._delete_container_for_test(container_id) + def test_change_feed_returns_cosmos_item_paged_with_etag(self): + """query_items_change_feed must return the public CosmosItemPaged type and expose the + change feed continuation token (etag) through the thread-safe get_response_headers() API.""" + container_id = "test_headers_cf_etag_" + str(uuid.uuid4()) + created_collection = self._create_container_for_test(container_id, PartitionKey(path="/pk")) + try: + for i in range(5): + created_collection.create_item(body={"pk": "test", "id": f"item_{i}", "value": i}) + + change_feed = created_collection.query_items_change_feed(start_time="Beginning") + + # The returned pageable must be the public CosmosItemPaged type + self.assertIsInstance(change_feed, CosmosItemPaged) + + items = list(change_feed) + self.assertEqual(len(items), 5) + + # Continuation token (etag) is available via the public thread-safe API + headers = change_feed.get_response_headers() + self.assertIsInstance(headers, CaseInsensitiveDict) + self.assertIn("etag", headers) + self.assertTrue(headers["etag"]) + finally: + self._delete_container_for_test(container_id) + + def test_change_feed_response_headers_empty(self): + """query_items_change_feed must still surface the continuation token (etag) via + get_response_headers() even when there are no changes to return.""" + container_id = "test_headers_cf_empty_" + str(uuid.uuid4()) + created_collection = self._create_container_for_test(container_id, PartitionKey(path="/pk")) + try: + # start_time="Now" with no writes -> empty change feed + change_feed = created_collection.query_items_change_feed(start_time="Now") + self.assertIsInstance(change_feed, CosmosItemPaged) + items = list(change_feed) + self.assertEqual(len(items), 0) + + # Continuation token should still be captured to allow resuming later + headers = change_feed.get_response_headers() + self.assertIn("etag", headers) + finally: + self._delete_container_for_test(container_id) + if __name__ == "__main__": unittest.main() diff --git a/sdk/cosmos/azure-cosmos/tests/test_query_response_headers_async.py b/sdk/cosmos/azure-cosmos/tests/test_query_response_headers_async.py index 96dfe2580f40..a8ffddbfb56b 100644 --- a/sdk/cosmos/azure-cosmos/tests/test_query_response_headers_async.py +++ b/sdk/cosmos/azure-cosmos/tests/test_query_response_headers_async.py @@ -917,6 +917,49 @@ def hook(headers, _body): finally: await self._delete_container_for_test(cid) + async def test_change_feed_returns_cosmos_item_paged_with_etag_async(self): + """query_items_change_feed must return the public CosmosAsyncItemPaged type and expose the + change feed continuation token (etag) through the thread-safe get_response_headers() API.""" + cid = "test_headers_cf_etag_async_" + str(uuid.uuid4()) + created_collection = await self._create_container_for_test(cid, PartitionKey(path="/pk")) + try: + for i in range(5): + await created_collection.create_item(body={"pk": "test", "id": f"item_{i}", "value": i}) + + change_feed = created_collection.query_items_change_feed(start_time="Beginning") + + # The returned pageable must be the public CosmosAsyncItemPaged type + self.assertIsInstance(change_feed, CosmosAsyncItemPaged) + + items = [item async for item in change_feed] + self.assertEqual(len(items), 5) + + # Continuation token (etag) is available via the public thread-safe API + headers = change_feed.get_response_headers() + self.assertIsInstance(headers, CaseInsensitiveDict) + self.assertIn("etag", headers) + self.assertTrue(headers["etag"]) + finally: + await self._delete_container_for_test(cid) + + async def test_change_feed_response_headers_empty_async(self): + """query_items_change_feed must still surface the continuation token (etag) via + get_response_headers() even when there are no changes to return.""" + cid = "test_headers_cf_empty_async_" + str(uuid.uuid4()) + created_collection = await self._create_container_for_test(cid, PartitionKey(path="/pk")) + try: + # start_time="Now" with no writes -> empty change feed + change_feed = created_collection.query_items_change_feed(start_time="Now") + self.assertIsInstance(change_feed, CosmosAsyncItemPaged) + items = [item async for item in change_feed] + self.assertEqual(len(items), 0) + + # Continuation token should still be captured to allow resuming later + headers = change_feed.get_response_headers() + self.assertIn("etag", headers) + finally: + await self._delete_container_for_test(cid) + if __name__ == "__main__": unittest.main()