Create itemupdate model; create unittest for new methods in stac item

datacosmos/client.py

@@ -110,6 +110,10 @@ def post(self, url: str, *args: Any, **kwargs: Any) -> requests.Response:
     def put(self, url: str, *args: Any, **kwargs: Any) -> requests.Response:
         """Send a PUT request using the authenticated session."""
         return self.request("PUT", url, *args, **kwargs)
+
+    def patch(self, url: str, *args: Any, **kwargs: Any) -> requests.Response:
+        """Send a PATCH request using the authenticated session."""
+        return self.request("PATCH", url, *args, **kwargs)
 
     def delete(self, url: str, *args: Any, **kwargs: Any) -> requests.Response:
         """Send a DELETE request using the authenticated session."""

datacosmos/stac/models/item_update.py

@@ -0,0 +1,22 @@
+from typing import Optional, Any
+from pydantic import BaseModel, Field
+from pystac import Asset, Link
+from shapely.geometry import mapping
+
+
+class ItemUpdate(BaseModel):
+    """Model representing a partial update for a STAC item."""
+    model_config = {
+        "arbitrary_types_allowed": True
+    }
+
+    stac_extensions: Optional[list[str]] = None
+    geometry: Optional[dict[str, Any]] = None
+    bbox: Optional[list[float]] = Field(None, min_items=4, max_items=4)  # Must be [minX, minY, maxX, maxY]
+    properties: Optional[dict[str, Any]] = None
+    assets: Optional[dict[str, Asset]] = None
+    links: Optional[list[Link]] = None
+
+    def set_geometry(self, geom) -> None:
+        """Convert a shapely geometry to GeoJSON format."""
+        self.geometry = mapping(geom)

datacosmos/stac/stac_client.py

@@ -9,7 +9,8 @@
 from pystac import Item
 from datacosmos.client import DatacosmosClient
 from datacosmos.stac.models.search_parameters import SearchParameters
-from common.sdk.http_response import check_api_response
+from datacosmos.stac.models.item_update import ItemUpdate
+from common.sdk.http_response import check_api_response, InvalidRequest
 
 class STACClient:
     """Client for interacting with the STAC API."""
@@ -84,28 +85,39 @@ def create_item(self, collection_id: str, item: Item) -> Item:
             RequestError: If the API returns an error response.
         """
         url = self.base_url.with_suffix(f"/collections/{collection_id}/items")
-        item_json: dict = item.to_dict()  # Convert the STAC Item to JSON format
+        item_json: dict = item.to_dict()
 
         response = self.client.post(url, json=item_json)
         check_api_response(response)
 
         return Item.from_dict(response.json())
 
-    def update_item(self, item_id: str, collection_id: str, update_data: dict) -> Item:
-        """Update an existing STAC item.
+    def update_item(self, item_id: str, collection_id: str, update_data: ItemUpdate) -> Item:
+        """Partially update an existing STAC item.
 
         Args:
             item_id (str): The ID of the item to update.
             collection_id (str): The ID of the collection containing the item.
-            update_data (dict): The update data (partial or full).
+            update_data (ItemUpdate): The structured update payload.
 
         Returns:
             Item: The updated STAC item.
         """
         url = self.base_url.with_suffix(f"/collections/{collection_id}/items/{item_id}")
-        response = self.client.patch(url, json=update_data)
+
+        update_payload = update_data.model_dump(by_alias=True, exclude_none=True)
+
+        if "assets" in update_payload:
+            update_payload["assets"] = {
+                key: asset.to_dict() for key, asset in update_payload["assets"].items()
+            }
+        if "links" in update_payload:
+            update_payload["links"] = [link.to_dict() for link in update_payload["links"]]
+
+        response = self.client.patch(url, json=update_payload)
         check_api_response(response)
-        return Item.from_dict(response.json())
+
+        return Item.from_dict(response.json()) 
 
     def delete_item(self, item_id: str, collection_id: str) -> None:
         """Delete a STAC item by its ID.
@@ -157,9 +169,18 @@ def _get_next_link(self, data: dict) -> Optional[str]:
         return next_link.get("href", "") if next_link else None
 
     def _extract_pagination_token(self, next_href: str) -> Optional[str]:
-        """Extract the pagination token from the next link URL."""
+        """Extract the pagination token from the next link URL.
+
+        Args:
+            next_href (str): The next page URL.
+
+        Returns:
+            Optional[str]: The extracted token, or None if parsing fails.
+
+        Raises:
+            InvalidRequest: If pagination token extraction fails.
+        """
         try:
             return next_href.split("?")[1].split("=")[-1]
         except (IndexError, AttributeError):
-            self.client.logger.error(f"Failed to parse pagination token from {next_href}")
-            return None
+            raise InvalidRequest(f"Failed to parse pagination token from {next_href}")

tests/unit/datacosmos/client/test_client_patch_request.py

@@ -0,0 +1,40 @@
+from unittest.mock import MagicMock, patch
+
+from config.config import Config
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
+from datacosmos.client import DatacosmosClient
+
+
+@patch("datacosmos.client.DatacosmosClient._authenticate_and_initialize_client")
+def test_patch_request(mock_auth_client):
+    """Test that the client performs a PATCH request correctly."""
+    # Mock the HTTP client
+    mock_http_client = MagicMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"message": "updated"}
+    mock_http_client.request.return_value = mock_response
+    mock_auth_client.return_value = mock_http_client
+
+    config = Config(
+        authentication=M2MAuthenticationConfig(
+            type="m2m",
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_url="https://mock.token.url/oauth/token",
+            audience="https://mock.audience",
+        )
+    )
+
+    client = DatacosmosClient(config=config)
+    response = client.patch(
+        "https://mock.api/some-endpoint", json={"key": "updated-value"}
+    )
+
+    # Assertions
+    assert response.status_code == 200
+    assert response.json() == {"message": "updated"}
+    mock_http_client.request.assert_called_once_with(
+        "PATCH", "https://mock.api/some-endpoint", json={"key": "updated-value"}
+    )
+    mock_auth_client.call_count == 2

tests/unit/datacosmos/stac/stac_client/test_create_item.py

@@ -0,0 +1,49 @@
+from unittest.mock import MagicMock, patch
+from pystac import Item
+
+from config.config import Config
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
+from datacosmos.client import DatacosmosClient
+from datacosmos.stac.stac_client import STACClient
+
+
+@patch("requests_oauthlib.OAuth2Session.fetch_token")
+@patch.object(DatacosmosClient, "post")
+@patch("datacosmos.stac.stac_client.check_api_response")
+def test_create_item(mock_check_api_response, mock_post, mock_fetch_token):
+    """Test creating a new STAC item."""
+    mock_fetch_token.return_value = {"access_token": "mock-token", "expires_in": 3600}
+
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "id": "item-1",
+        "collection": "test-collection",
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "geometry": {"type": "Point", "coordinates": [0, 0]},
+        "properties": {"datetime": "2023-12-01T12:00:00Z"},
+        "assets": {},
+        "links": [],
+    }
+    mock_post.return_value = mock_response
+
+    config = Config(
+        authentication=M2MAuthenticationConfig(
+            type="m2m",
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_url="https://mock.token.url/oauth/token",
+            audience="https://mock.audience",
+        )
+    )
+
+    client = DatacosmosClient(config=config)
+    stac_client = STACClient(client)
+
+    item = Item.from_dict(mock_response.json())
+    created_item = stac_client.create_item("test-collection", item)
+
+    assert created_item.id == "item-1"
+    assert created_item.properties["datetime"] == "2023-12-01T12:00:00Z"
+    mock_post.assert_called_once()
+    mock_check_api_response.assert_called_once()

tests/unit/datacosmos/stac/stac_client/test_delete_item.py

@@ -0,0 +1,36 @@
+from unittest.mock import MagicMock, patch
+
+from config.config import Config
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
+from datacosmos.client import DatacosmosClient
+from datacosmos.stac.stac_client import STACClient
+
+
+@patch("requests_oauthlib.OAuth2Session.fetch_token")
+@patch.object(DatacosmosClient, "delete")
+@patch("datacosmos.stac.stac_client.check_api_response")
+def test_delete_item(mock_check_api_response, mock_delete, mock_fetch_token):
+    """Test deleting a STAC item."""
+    mock_fetch_token.return_value = {"access_token": "mock-token", "expires_in": 3600}
+
+    mock_response = MagicMock()
+    mock_response.status_code = 204  # Successful deletion
+    mock_delete.return_value = mock_response
+
+    config = Config(
+        authentication=M2MAuthenticationConfig(
+            type="m2m",
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_url="https://mock.token.url/oauth/token",
+            audience="https://mock.audience",
+        )
+    )
+
+    client = DatacosmosClient(config=config)
+    stac_client = STACClient(client)
+
+    stac_client.delete_item("item-1", "test-collection")
+
+    mock_delete.assert_called_once()
+    mock_check_api_response.assert_called_once()

tests/unit/datacosmos/stac/stac_client/test_update_item.py

@@ -0,0 +1,50 @@
+from unittest.mock import MagicMock, patch
+from pystac import Item
+from datacosmos.stac.models.item_update import ItemUpdate
+
+from config.config import Config
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
+from datacosmos.client import DatacosmosClient
+from datacosmos.stac.stac_client import STACClient
+
+
+@patch("requests_oauthlib.OAuth2Session.fetch_token")
+@patch.object(DatacosmosClient, "patch")
+@patch("datacosmos.stac.stac_client.check_api_response")
+def test_update_item(mock_check_api_response, mock_patch, mock_fetch_token):
+    """Test updating an existing STAC item."""
+    mock_fetch_token.return_value = {"access_token": "mock-token", "expires_in": 3600}
+
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "id": "item-1",
+        "collection": "test-collection",
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "geometry": {"type": "Point", "coordinates": [0, 0]},
+        "properties": {"datetime": "2023-12-01T12:00:00Z", "new_property": "value"},
+        "assets": {},
+        "links": [],
+    }
+    mock_patch.return_value = mock_response
+
+    config = Config(
+        authentication=M2MAuthenticationConfig(
+            type="m2m",
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_url="https://mock.token.url/oauth/token",
+            audience="https://mock.audience",
+        )
+    )
+
+    client = DatacosmosClient(config=config)
+    stac_client = STACClient(client)
+
+    update_data = ItemUpdate(properties={"new_property": "value"})
+    updated_item = stac_client.update_item("item-1", "test-collection", update_data)
+
+    assert updated_item.id == "item-1"
+    assert updated_item.properties["new_property"] == "value"
+    mock_patch.assert_called_once()
+    mock_check_api_response.assert_called_once()