feat(assets): add asset history API methods

pythonik/models/base.py

@@ -41,6 +41,44 @@ class StorageMethod(str, Enum):
     GCS = "GCS"
 
 
+class HistoryOperationType(str, Enum):
+    """Known asset history operation types.
+
+    This enum is provided for convenience and discoverability. The Iconik API
+    may add new operation types over time, so ``create_history_entity`` also
+    accepts arbitrary strings rather than restricting callers to these values.
+    """
+
+    EXPORT = "EXPORT"
+    TRANSCODE = "TRANSCODE"
+    ANALYZE = "ANALYZE"
+    ADD_FORMAT = "ADD_FORMAT"
+    DELETE_FORMAT = "DELETE_FORMAT"
+    RESTORE_FORMAT = "RESTORE_FORMAT"
+    DELETE_FILESET = "DELETE_FILESET"
+    DELETE_FILE = "DELETE_FILE"
+    RESTORE_FILESET = "RESTORE_FILESET"
+    MODIFY_FILESET = "MODIFY_FILESET"
+    APPROVE = "APPROVE"
+    REJECT = "REJECT"
+    DOWNLOAD = "DOWNLOAD"
+    METADATA = "METADATA"
+    CUSTOM = "CUSTOM"
+    TRANSCRIPTION = "TRANSCRIPTION"
+    VERSION_CREATE = "VERSION_CREATE"
+    VERSION_DELETE = "VERSION_DELETE"
+    VERSION_UPDATE = "VERSION_UPDATE"
+    VERSION_PROMOTE = "VERSION_PROMOTE"
+    RESTORE = "RESTORE"
+    RESTORE_FROM_GLACIER = "RESTORE_FROM_GLACIER"
+    ARCHIVE = "ARCHIVE"
+    RESTORE_ARCHIVE = "RESTORE_ARCHIVE"
+    DELETE = "DELETE"
+    TRANSFER = "TRANSFER"
+    UNLINK_SUBCLIP = "UNLINK_SUBCLIP"
+    FACE_RECOGNITION = "FACE_RECOGNITION"
+
+
 class PaginatedResponse(BaseModel):
     first_url: Optional[str] = ""
     last_url: Optional[str] = ""

pythonik/specs/assets.py

@@ -14,13 +14,19 @@
     AssetVersionFromAssetCreate,
     AssetVersion,
 )
-from pythonik.models.base import Response
+from pythonik.models.base import (
+    Response,
+    PaginatedResponse,
+    HistoryOperationType,
+)
 from pythonik.specs.base import Spec
 from pythonik.specs.collection import CollectionSpec
 
 BASE = "assets"
 DELETE_QUEUE = "delete_queue"
+LIST_URL = BASE + "/"
 GET_URL = BASE + "/{}/"
+HISTORY_URL = BASE + "/{}/history/"
 SEGMENT_URL = BASE + "/{}/segments/"
 GET_SEGMENTS_URL = BASE + "/{}/segments/"
 SEGMENT_URL_UPDATE = SEGMENT_URL + "{}/"
@@ -610,3 +616,61 @@ def get_segments(
 
         response = self._get(GET_SEGMENTS_URL.format(asset_id), params=params, **kwargs)
         return self.parse_response(response, SegmentListResponse)
+
+    def list_all(self, **kwargs) -> Response:
+        """
+        Get list of assets.
+
+        Args:
+            **kwargs: Additional kwargs to pass to the request
+
+        Returns:
+            Response(model=PaginatedResponse) containing paginated asset list
+        """
+        resp = self._get(LIST_URL, **kwargs)
+        return self.parse_response(resp, PaginatedResponse)
+
+    def list_asset_history_entities(self, asset_id: str, **kwargs) -> Response:
+        """
+        Get list of history entities for asset.
+
+        Args:
+            asset_id: ID of the asset
+            **kwargs: Additional kwargs to pass to the request
+
+        Returns:
+            Response(model=PaginatedResponse) containing history entities
+        """
+        resp = self._get(HISTORY_URL.format(asset_id), **kwargs)
+        return self.parse_response(resp, PaginatedResponse)
+
+    def create_history_entity(
+            self, asset_id: str,
+            operation_description: str,
+            operation_type: Union[HistoryOperationType, str],
+            **kwargs
+    ) -> Response:
+        """
+        Create an asset history entity.
+
+        Args:
+            asset_id: ID of the asset
+            operation_description: Description of the operation
+            operation_type: Type of operation. Prefer a
+                :class:`~pythonik.models.base.HistoryOperationType` member for
+                the known values (e.g. ``HistoryOperationType.VERSION_CREATE``),
+                but any string is accepted so newly added API operation types
+                keep working without an SDK upgrade.
+            **kwargs: Additional kwargs to pass to the request
+
+        Returns:
+            Response with history entry creation status
+        """
+        if isinstance(operation_type, HistoryOperationType):
+            operation_type = operation_type.value
+        body = {
+            "operation_description": operation_description,
+            "operation_type": operation_type
+        }
+        resp = self._post(HISTORY_URL.format(asset_id), json=body, **kwargs)
+        return self.parse_response(resp, None)

pythonik/tests/test_assets_history.py

@@ -0,0 +1,210 @@
+# pythonik/tests/test_assets_history.py
+import uuid
+import requests_mock
+
+from pythonik.client import PythonikClient
+from pythonik.models.base import PaginatedResponse, HistoryOperationType
+from pythonik.specs.assets import AssetSpec
+
+
+def test_list_all():
+    """Test fetching a list of assets."""
+    with requests_mock.Mocker() as m:
+        app_id = str(uuid.uuid4())
+        auth_token = str(uuid.uuid4())
+
+        # Mock response data structure
+        response_data = {
+            "objects": [
+                {"id": str(uuid.uuid4()), "title": "Test Asset 1", "status": "ACTIVE"},
+                {"id": str(uuid.uuid4()), "title": "Test Asset 2", "status": "ACTIVE"},
+            ],
+            "page": 1,
+            "pages": 1,
+            "per_page": 2,
+            "total": 2,
+        }
+
+        # Setup mock endpoint
+        mock_address = AssetSpec.gen_url("assets/")
+        m.get(mock_address, json=response_data)
+
+        # Create client and call the method
+        client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3)
+        result = client.assets().list_all()
+
+        # Verify response
+        assert result.response.ok
+        assert isinstance(result.data, PaginatedResponse)
+        assert len(result.data.objects) == 2
+        assert result.data.page == 1
+        assert result.data.pages == 1
+        assert result.data.total == 2
+
+
+def test_list_all_with_params():
+    """Test fetching a list of assets with query parameters."""
+    with requests_mock.Mocker() as m:
+        app_id = str(uuid.uuid4())
+        auth_token = str(uuid.uuid4())
+
+        # Mock response data
+        response_data = {
+            "objects": [
+                {"id": str(uuid.uuid4()), "title": "Test Asset 1", "status": "ACTIVE"}
+            ],
+            "page": 1,
+            "pages": 1,
+            "per_page": 1,
+            "total": 1,
+        }
+
+        # Setup mock endpoint with params matcher
+        mock_address = AssetSpec.gen_url("assets/")
+        m.get(
+            mock_address,
+            json=response_data,
+            # Add request matcher to ensure params are passed correctly
+            additional_matcher=lambda req: req.qs == {"page": ["1"], "per_page": ["1"]},
+        )
+
+        # Create client and call method with params
+        client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3)
+        params = {"page": 1, "per_page": 1}
+        result = client.assets().list_all(params=params)
+
+        # Verify response
+        assert result.response.ok
+        assert isinstance(result.data, PaginatedResponse)
+        assert len(result.data.objects) == 1
+
+
+def test_list_asset_history_entities():
+    """Test fetching history entities for an asset."""
+    with requests_mock.Mocker() as m:
+        app_id = str(uuid.uuid4())
+        auth_token = str(uuid.uuid4())
+        asset_id = str(uuid.uuid4())
+
+        # Mock response data
+        response_data = {
+            "objects": [
+                {
+                    "id": str(uuid.uuid4()),
+                    "operation_type": "METADATA",
+                    "operation_description": "Updated metadata",
+                    "date_created": "2025-05-13T10:00:00Z",
+                    "created_by_user": "user123",
+                },
+                {
+                    "id": str(uuid.uuid4()),
+                    "operation_type": "VERSION_CREATE",
+                    "operation_description": "Created new version",
+                    "date_created": "2025-05-12T15:30:00Z",
+                    "created_by_user": "user123",
+                },
+            ],
+            "page": 1,
+            "pages": 1,
+            "per_page": 10,
+            "total": 2,
+        }
+
+        # Setup mock endpoint
+        mock_address = AssetSpec.gen_url(f"assets/{asset_id}/history/")
+        m.get(mock_address, json=response_data)
+
+        # Create client and call the method
+        client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3)
+        result = client.assets().list_asset_history_entities(asset_id)
+
+        # Verify response
+        assert result.response.ok
+        assert isinstance(result.data, PaginatedResponse)
+        assert len(result.data.objects) == 2
+        assert result.data.objects[0]["operation_type"] == "METADATA"
+        assert result.data.objects[1]["operation_type"] == "VERSION_CREATE"
+
+
+def test_create_history_entity():
+    """Test creating a history entity for an asset."""
+    with requests_mock.Mocker() as m:
+        app_id = str(uuid.uuid4())
+        auth_token = str(uuid.uuid4())
+        asset_id = str(uuid.uuid4())
+        operation_description = "Test history entry"
+        operation_type = "CUSTOM"
+
+        # Setup mock endpoint
+        mock_address = AssetSpec.gen_url(f"assets/{asset_id}/history/")
+        m.post(mock_address, status_code=201)
+
+        # Create client and call the method
+        client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3)
+        result = client.assets().create_history_entity(
+            asset_id=asset_id,
+            operation_description=operation_description,
+            operation_type=operation_type,
+        )
+
+        # Verify response
+        assert result.response.ok
+        assert result.response.status_code == 201
+
+        # Verify the correct request body was sent
+        assert m.last_request.json() == {
+            "operation_description": operation_description,
+            "operation_type": operation_type,
+        }
+
+
+def test_create_history_entity_accepts_enum():
+    """Passing a HistoryOperationType enum sends its string value."""
+    with requests_mock.Mocker() as m:
+        app_id = str(uuid.uuid4())
+        auth_token = str(uuid.uuid4())
+        asset_id = str(uuid.uuid4())
+        operation_description = "Created new version"
+
+        mock_address = AssetSpec.gen_url(f"assets/{asset_id}/history/")
+        m.post(mock_address, status_code=201)
+
+        client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3)
+        result = client.assets().create_history_entity(
+            asset_id=asset_id,
+            operation_description=operation_description,
+            operation_type=HistoryOperationType.VERSION_CREATE,
+        )
+
+        assert result.response.ok
+        # Enum is serialized to its plain string value in the request body.
+        assert m.last_request.json() == {
+            "operation_description": operation_description,
+            "operation_type": "VERSION_CREATE",
+        }
+
+
+def test_create_history_entity_accepts_arbitrary_string():
+    """Unknown operation types are accepted (API may add new ones)."""
+    with requests_mock.Mocker() as m:
+        app_id = str(uuid.uuid4())
+        auth_token = str(uuid.uuid4())
+        asset_id = str(uuid.uuid4())
+        operation_description = "Some brand new operation"
+        operation_type = "SOME_FUTURE_OPERATION"
+
+        mock_address = AssetSpec.gen_url(f"assets/{asset_id}/history/")
+        m.post(mock_address, status_code=201)
+
+        client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3)
+        result = client.assets().create_history_entity(
+            asset_id=asset_id,
+            operation_description=operation_description,
+            operation_type=operation_type,
+        )
+
+        assert result.response.ok
+        assert m.last_request.json() == {
+            "operation_description": operation_description,
+            "operation_type": operation_type,
+        }