Added Response Headers to Control Plane Operations

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py

@@ -93,6 +93,7 @@ def __init__(
         self.container_link = "{}/colls/{}".format(database_link, self.id)
         self._is_system_key: Optional[bool] = None
         self._scripts: Optional[ScriptsProxy] = None
+        self._response_headers = properties
         if properties:
             self.client_connection._set_container_properties_cache(self.container_link,
                                                                    _build_properties_cache(properties,
@@ -112,6 +113,14 @@ async def _get_properties(self, **kwargs: Any) -> Dict[str, Any]:
             await self.read(**kwargs)
         return self.client_connection._container_properties_cache[self.container_link]
 
+    def get_response_headers(self) -> dict[str, Any]:
+        """Returns a copy of the response headers associated to this response
+
+        :return: Dict of response headers
+        :rtype: dict[str, Any]
+        """
+        return self._response_headers.copy()
+
     @property
     async def is_system_key(self) -> bool:
         if self._is_system_key is None:

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_database.py

@@ -100,6 +100,7 @@ def __init__(
         self.id = id
         self.database_link = "dbs/{}".format(self.id)
         self._properties = properties
+        self._response_headers = properties
 
     def __repr__(self) -> str:
         return "<DatabaseProxy [{}]>".format(self.database_link)[:1024]
@@ -126,6 +127,14 @@ async def _get_properties(self) -> Dict[str, Any]:
             self._properties = await self.read()
         return self._properties
 
+    def get_response_headers(self) -> dict[str, Any]:
+        """Returns a copy of the response headers associated to this response
+
+        :return: Dict of response headers
+        :rtype: dict[str, Any]
+        """
+        return self._response_headers.copy()
+
     @distributed_trace_async
     async def read(
         self,

sdk/cosmos/azure-cosmos/azure/cosmos/container.py

@@ -104,6 +104,7 @@ def __init__(
         self.client_connection = client_connection
         self._is_system_key: Optional[bool] = None
         self._scripts: Optional[ScriptsProxy] = None
+        self._response_headers = properties
         if properties:
             self.client_connection._set_container_properties_cache(self.container_link,
                                                                    _build_properties_cache(properties,
@@ -123,6 +124,14 @@ def _get_properties(self, **kwargs: Any) -> Dict[str, Any]:
             self.read(**kwargs)
         return self.__get_client_container_caches()[self.container_link]
 
+    def get_response_headers(self) -> dict[str, Any]:
+        """Returns a copy of the response headers associated to this response
+
+        :return: Dict of response headers
+        :rtype: dict[str, Any]
+        """
+        return self._response_headers.copy()
+
     @property
     def is_system_key(self) -> bool:
         if self._is_system_key is None:

sdk/cosmos/azure-cosmos/azure/cosmos/database.py

@@ -94,6 +94,7 @@ def __init__(
         self.id = id
         self.database_link: str = "dbs/{}".format(self.id)
         self._properties: Optional[Dict[str, Any]] = properties
+        self._response_headers = properties
 
     def __repr__(self) -> str:
         return "<DatabaseProxy [{}]>".format(self.database_link)[:1024]
@@ -120,6 +121,14 @@ def _get_properties(self) -> Dict[str, Any]:
             self._properties = self.read()
         return self._properties
 
+    def get_response_headers(self) -> dict[str, Any]:
+        """Returns a copy of the response headers associated to this response
+
+        :return: Dict of response headers
+        :rtype: dict[str, Any]
+        """
+        return self._response_headers.copy()
+
     @distributed_trace
     def read(  # pylint:disable=docstring-missing-param
         self,

sdk/cosmos/azure-cosmos/tests/test_cosmos_responses.py

@@ -69,6 +69,30 @@ def test_point_operation_headers(self):
         assert len(batch_response.get_response_headers()) > 0
         assert int(lsn) + 1 < int(batch_response.get_response_headers()['lsn'])
 
+    def test_create_database_headers(self):
+        first_response = self.client.create_database(id="responses_test" + str(uuid.uuid4()))
+        assert len(first_response.get_response_headers()) > 0
+
+    def test_create_database_if_not_exists_headers(self):
+        first_response = self.client.create_database_if_not_exists(id="responses_test" + str(uuid.uuid4()))
+        assert len(first_response.get_response_headers()) > 0
+
+    def test_create_container_headers(self):
+        first_response = self.test_database.create_container(id="responses_test" + str(uuid.uuid4()),
+                                                        partition_key=PartitionKey(path="/company"))
+        assert len(first_response.get_response_headers()) > 0
+
+    def test_create_container_if_not_exists_headers(self):
+        first_response = self.test_database.create_container_if_not_exists(id="responses_test" + str(uuid.uuid4()),
+                                                        partition_key=PartitionKey(path="/company"))
+        assert len(first_response.get_response_headers()) > 0
+
+    def test_replace_container_headers(self):
+        first_response = self.test_database.create_container_if_not_exists(id="responses_test" + str(uuid.uuid4()),
+                                                        partition_key=PartitionKey(path="/company"))
+        second_response = self.test_database.replace_container(first_response.id,
+                                                               partition_key=PartitionKey(path="/company"))
+        assert len(second_response.get_response_headers()) > 0
 
 if __name__ == '__main__':
     unittest.main()

sdk/cosmos/azure-cosmos/tests/test_cosmos_responses_async.py

@@ -73,6 +73,30 @@ async def test_point_operation_headers_async(self):
         assert len(batch_response.get_response_headers()) > 0
         assert int(lsn) + 1 < int(batch_response.get_response_headers()['lsn'])
 
+    async def test_create_database_headers(self):
+        first_response = await self.client.create_database(id="responses_test" + str(uuid.uuid4()))
+        assert len(first_response.get_response_headers()) > 0
+
+    async def test_create_database_if_not_exists_headers(self):
+        first_response = await self.client.create_database_if_not_exists(id="responses_test" + str(uuid.uuid4()))
+        assert len(first_response.get_response_headers()) > 0
+
+    async def test_create_container_headers(self):
+        first_response = await self.test_database.create_container(id="responses_test" + str(uuid.uuid4()),
+                                                        partition_key=PartitionKey(path="/company"))
+        assert len(first_response.get_response_headers()) > 0
+
+    async def test_create_container_if_not_exists_headers(self):
+        first_response = await self.test_database.create_container_if_not_exists(id="responses_test" + str(uuid.uuid4()),
+                                                        partition_key=PartitionKey(path="/company"))
+        assert len(first_response.get_response_headers()) > 0
+
+    async def test_replace_container_headers(self):
+        first_response = await self.test_database.create_container_if_not_exists(id="responses_test" + str(uuid.uuid4()),
+                                                        partition_key=PartitionKey(path="/company"))
+        second_response = await self.test_database.replace_container(first_response.id,
+                                                               partition_key=PartitionKey(path="/company"))
+        assert len(second_response.get_response_headers()) > 0
 
 if __name__ == '__main__':
     unittest.main()