Skip to content

Return CosmosItemPaged from query_items_change_feed for thread-safe headers #47947

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andrewmathew1 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Return CosmosItemPaged from query_items_change_feed for thread-safe headers #47947

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions sdk/cosmos/azure-cosmos/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
25 changes: 17 additions & 8 deletions sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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

"""

Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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(
Expand Down
30 changes: 15 additions & 15 deletions sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
"""
...

Expand All @@ -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.
Expand Down Expand Up @@ -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
"""
...

Expand All @@ -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.
Expand All @@ -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
...
Expand All @@ -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.

Expand Down Expand Up @@ -1152,16 +1152,16 @@ 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
"""
...

@distributed_trace
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.

Expand Down Expand Up @@ -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)
Expand Down
25 changes: 17 additions & 8 deletions sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2459,17 +2459,19 @@ 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.
:param dict options: The request options for the request.
: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

"""

Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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)
Expand All @@ -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(
Expand Down
30 changes: 15 additions & 15 deletions sdk/cosmos/azure-cosmos/azure/cosmos/container.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
"""
...

Expand All @@ -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.

Expand Down Expand Up @@ -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
"""
...

Expand All @@ -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.
Expand All @@ -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
"""
...

Expand All @@ -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,

Expand Down Expand Up @@ -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
"""
...

Expand All @@ -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.
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading