diff --git a/elytra_client/rest_api/client.py b/elytra_client/rest_api/client.py index f24b8a2..718af94 100644 --- a/elytra_client/rest_api/client.py +++ b/elytra_client/rest_api/client.py @@ -14,19 +14,31 @@ from ..exceptions import ( ) from .auth import AuthMethod, RestApiAuth from .models import ( + AttributeGroupHierarchyResponse, + AttributeResponse, JobControlRequest, JobControlResponse, JobDetailInfo, JobExecutionResponse, JobInfo, JobOverviewResponse, + MediaBulkCreateResponse, + MediaBulkUpdateResponse, + MediaFileResponse, + MediaListResponse, + ProductGroupHierarchyResponse, + ProductHierarchyResponse, ProtocolCategoryInfo, ProtocolCategoryListResponse, ProtocolInfo, ProtocolListResponse, + SingleMediaResponse, + SingleNewMediaRequestBody, + SingleUpdateMediaRequestBody, + TreeGroupHierarchyResponse, ) -T = TypeVar('T', bound=BaseModel) +T = TypeVar("T", bound=BaseModel) class LobsterRestApiClient: @@ -425,6 +437,314 @@ class LobsterRestApiClient: params={"limit": limit}, ) + # ============= Media Endpoints ============= + + def get_all_media( + self, lang: Optional[str] = None, page: int = 1, limit: int = 10 + ) -> MediaListResponse: + """ + Get all media descriptors with pagination. + + Args: + lang: Language code (optional) + page: Page number (default: 1) + limit: Number of media items per page (default: 10) + + Returns: + MediaListResponse containing paginated list of media items + """ + params: Dict[str, Any] = {"page": page, "limit": limit} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + "media", + MediaListResponse, + params=params, + ) + + def get_media_by_id(self, media_id: int, lang: Optional[str] = None) -> SingleMediaResponse: + """ + Get a media descriptor by ID. + + Args: + media_id: ID of the media + lang: Language code (optional) + + Returns: + SingleMediaResponse with media details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"media/{media_id}", + SingleMediaResponse, + params=params if params else None, + ) + + def create_media(self, media_data: Dict[str, Any]) -> SingleMediaResponse: + """ + Create a new media descriptor. + + Args: + media_data: Media descriptor data + + Returns: + SingleMediaResponse with created media + """ + return self._make_request( + "POST", + "media", + SingleMediaResponse, + json_data=media_data, + ) + + def update_media(self, media_data: Dict[str, Any]) -> SingleMediaResponse: + """ + Update a media descriptor. + + Args: + media_data: Updated media descriptor data (must include 'id') + + Returns: + SingleMediaResponse with updated media + """ + return self._make_request( + "PATCH", + "media", + SingleMediaResponse, + json_data=media_data, + ) + + def create_multiple_media(self, media_list: List[Dict[str, Any]]) -> MediaBulkCreateResponse: + """ + Create multiple media descriptors in bulk. + + Args: + media_list: List of media descriptor data + + Returns: + MediaBulkCreateResponse with created media items + """ + return self._make_request( + "POST", + "media/bulk", + MediaBulkCreateResponse, + json_data=media_list, + ) + + def update_multiple_media(self, media_list: List[Dict[str, Any]]) -> MediaBulkUpdateResponse: + """ + Update multiple media descriptors in bulk. + + Args: + media_list: List of media descriptor data to update (each must include 'id') + + Returns: + MediaBulkUpdateResponse with updated media items + """ + return self._make_request( + "PATCH", + "media/bulk", + MediaBulkUpdateResponse, + json_data=media_list, + ) + + def delete_media(self, media_id: int) -> None: + """ + Delete a media descriptor by ID. + + Args: + media_id: ID of the media to delete + """ + url = urljoin(self.base_url, f"/rest/media/{media_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete media: {response.status_code}") + + def upload_media_file( + self, + file_path: str, + media_id: int, + mam_system: str, + language_code: str = "independent", + ) -> MediaFileResponse: + """ + Upload a media file for a media descriptor. + + Args: + file_path: Path to the file to upload + media_id: ID of the media descriptor + mam_system: MAM system code (fs, sixomc, cumulus, etc.) + language_code: Language code for the file (default: independent) + + Returns: + MediaFileResponse with uploaded file metadata + """ + url = urljoin(self.base_url, "/rest/media/file") + + with open(file_path, "rb") as file: + files = {"file": file} + data = { + "mediaId": media_id, + "mamSystem": mam_system, + "languageCode": language_code, + } + # Add authentication credentials + data.update(self.auth.get_json_body_params()) + + response = self.session.post( + url, + files=files, + data=data, + timeout=self.timeout, + ) + + return self._handle_response(response, MediaFileResponse) + + def get_media_content(self, media_file_id: int) -> bytes: + """ + Download the binary content of a media file. + + Args: + media_file_id: ID of the media file + + Returns: + Binary content of the media file + """ + url = urljoin(self.base_url, f"/rest/media/file/{media_file_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.get(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to download media: {response.status_code}") + + return response.content + + def delete_media_file(self, media_file_id: int) -> None: + """ + Delete a media file by ID. + + Args: + media_file_id: ID of the media file to delete + """ + url = urljoin(self.base_url, f"/rest/media/file/{media_file_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete media file: {response.status_code}") + + # ============= Hierarchy Endpoints ============= + + def get_product_hierarchy(self, product_id: int, depth: int = 10) -> ProductHierarchyResponse: + """ + Get the hierarchy of a product. + + Args: + product_id: ID of the product + depth: Depth of the hierarchy (default: 10) + + Returns: + ProductHierarchyResponse with hierarchical product structure + """ + params = {"depth": depth} + return self._make_request( + "GET", + f"products/{product_id}/hierarchy", + ProductHierarchyResponse, + params=params, + ) + + def get_product_group_hierarchy( + self, group_id: int, depth: int = 10 + ) -> ProductGroupHierarchyResponse: + """ + Get the hierarchy of a product group. + + Args: + group_id: ID of the product group + depth: Depth of the hierarchy (default: 10) + + Returns: + ProductGroupHierarchyResponse with hierarchical group structure + """ + params = {"depth": depth} + return self._make_request( + "GET", + f"groups/{group_id}/hierarchy", + ProductGroupHierarchyResponse, + params=params, + ) + + def get_tree_group_hierarchy(self, depth: int = 10) -> List[TreeGroupHierarchyResponse]: + """ + Get the hierarchy of tree groups. + + Args: + depth: Depth of the hierarchy (default: 10) + + Returns: + List of TreeGroupHierarchyResponse with hierarchical tree structure + """ + params = {"depth": depth} + url = urljoin(self.base_url, "/rest/tree/groups/hierarchy") + + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params.update(self.auth.get_url_parameters()) + + response = self.session.request( + method="GET", + url=url, + params=params, + timeout=self.timeout, + ) + + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to get tree hierarchy: {response.status_code}") + + try: + data = response.json() + if isinstance(data, list): + return [TreeGroupHierarchyResponse.model_validate(item) for item in data] + else: + return [TreeGroupHierarchyResponse.model_validate(data)] + except Exception as e: + raise ElytraValidationError(f"Failed to parse tree hierarchy: {str(e)}") + + def get_attribute_group_hierarchy( + self, attribute_group_id: int, depth: int = 10 + ) -> AttributeGroupHierarchyResponse: + """ + Get the hierarchy of an attribute group. + + Args: + attribute_group_id: ID of the attribute group + depth: Depth of the hierarchy (default: 10) + + Returns: + AttributeGroupHierarchyResponse with hierarchical attribute group structure + """ + params = {"depth": depth} + return self._make_request( + "GET", + f"attribute/groups/hierarchy/{attribute_group_id}", + AttributeGroupHierarchyResponse, + params=params, + ) + def close(self) -> None: """Close the session and clean up resources.""" self.session.close() diff --git a/elytra_client/rest_api/models.py b/elytra_client/rest_api/models.py index 0bbabe9..c82a370 100644 --- a/elytra_client/rest_api/models.py +++ b/elytra_client/rest_api/models.py @@ -7,6 +7,7 @@ from pydantic import BaseModel, Field class JobInfo(BaseModel): """Base job information model""" + id: int = Field(..., description="The ID of the job") name: str = Field(..., description="The name of the job") jobIdentifier: str = Field(..., description="The unique job identifier") @@ -22,12 +23,14 @@ class JobInfo(BaseModel): class JobDetailInfo(JobInfo): """Detailed job information including error level and runtime ID""" + errorLevel: Optional[str] = Field(None, description="Error level (e.g., 'Erfolgreich')") runtimeId: Optional[str] = Field(None, description="Runtime ID for active job execution") class JobOverviewResponse(BaseModel): """Response containing multiple job information items""" + jobInfoObjects: List[JobDetailInfo] = Field(..., description="List of job information objects") errors: List[str] = Field(default_factory=list, description="List of errors") warnings: List[str] = Field(default_factory=list, description="List of warnings") @@ -35,6 +38,7 @@ class JobOverviewResponse(BaseModel): class JobExecutionResponse(BaseModel): """Response from executing a job""" + id: int = Field(..., description="The ID of the job") name: str = Field(..., description="The name of the job") jobIdentifier: str = Field(..., description="The unique job identifier") @@ -44,12 +48,15 @@ class JobExecutionResponse(BaseModel): protocolId: str = Field(..., description="ID of the protocol for this execution") runtimeId: str = Field(..., description="Runtime ID for tracking execution") errors: List[str] = Field(default_factory=list, description="List of errors") - messages: List[str] = Field(default_factory=list, description="List of messages (e.g., JOB_START_OK)") + messages: List[str] = Field( + default_factory=list, description="List of messages (e.g., JOB_START_OK)" + ) warnings: List[str] = Field(default_factory=list, description="List of warnings") class JobControlRequest(BaseModel): """Request body for job control endpoint""" + action: str = Field(..., description="Action to perform (e.g., 'start')") objectId: int = Field(..., description="The ID of the job to control") objectType: str = Field(default="job", description="Type of object") @@ -61,9 +68,7 @@ class JobControlRequest(BaseModel): parameter: Optional[Dict[str, Any]] = Field( None, description="Parameters to override job settings" ) - queueId: Optional[str] = Field( - None, description="Queue ID for serialized job execution" - ) + queueId: Optional[str] = Field(None, description="Queue ID for serialized job execution") maxJobDurationSeconds: Optional[int] = Field( default=43200, description="Max duration in seconds (default 12 hours)" ) @@ -71,6 +76,7 @@ class JobControlRequest(BaseModel): class JobControlResponse(BaseModel): """Response from job control endpoint""" + jobIdentifier: str = Field(..., description="The job identifier") runtimeId: str = Field(..., description="Runtime ID for tracking") errors: List[str] = Field(default_factory=list, description="List of errors") @@ -80,6 +86,7 @@ class JobControlResponse(BaseModel): class ProtocolEntry(BaseModel): """A single entry in a protocol log""" + timestamp: Optional[str] = Field(None, description="Timestamp of the entry") level: Optional[str] = Field(None, description="Log level (ERROR, WARNING, INFO, etc.)") message: Optional[str] = Field(None, description="Message content") @@ -87,6 +94,7 @@ class ProtocolEntry(BaseModel): class ProtocolInfo(BaseModel): """Protocol/Log information""" + id: Optional[int] = Field(None, description="Protocol ID") protocolId: Optional[str] = Field(None, description="Protocol ID as string") jobId: Optional[int] = Field(None, description="Associated job ID") @@ -97,13 +105,12 @@ class ProtocolInfo(BaseModel): endTime: Optional[str] = Field(None, description="End time of execution") errors: List[str] = Field(default_factory=list, description="List of errors") messages: List[str] = Field(default_factory=list, description="List of messages") - entries: Optional[List[ProtocolEntry]] = Field( - None, description="Protocol entries" - ) + entries: Optional[List[ProtocolEntry]] = Field(None, description="Protocol entries") class ProtocolListResponse(BaseModel): """Response containing list of protocols""" + protocols: Optional[List[ProtocolInfo]] = Field(None, description="List of protocols") errors: List[str] = Field(default_factory=list, description="List of errors") warnings: List[str] = Field(default_factory=list, description="List of warnings") @@ -111,6 +118,7 @@ class ProtocolListResponse(BaseModel): class ProtocolCategoryInfo(BaseModel): """Protocol category information""" + id: str = Field(..., description="Category ID") name: str = Field(..., description="Category name") description: Optional[str] = Field(None, description="Category description") @@ -118,12 +126,243 @@ class ProtocolCategoryInfo(BaseModel): class ProtocolCategoryListResponse(BaseModel): """Response containing list of protocol categories""" + categories: List[ProtocolCategoryInfo] = Field(..., description="List of protocol categories") errors: List[str] = Field(default_factory=list, description="List of errors") class ErrorResponse(BaseModel): """Error response from the REST API""" + error: str = Field(..., description="Error message") errorCode: Optional[str] = Field(None, description="Error code") details: Optional[str] = Field(None, description="Error details") + + +# ============= Media and Hierarchy Models ============= + + +class PaginationLinks(BaseModel): + """Pagination links for list responses""" + + self: str = Field(..., description="Link to current page") + next: Optional[str] = Field(None, description="Link to next page") + previous: Optional[str] = Field(None, description="Link to previous page") + first: str = Field(..., description="Link to first page") + last: str = Field(..., description="Link to last page") + + +class AttributeResponse(BaseModel): + """Attribute value associated with an object""" + + id: int = Field(..., description="The ID of the attribute value") + attributeId: int = Field(..., description="The ID of the attribute definition") + attributeName: str = Field(..., description="The independent name of the attribute") + attributeType: str = Field( + ..., description="The category type of the attribute (normal, meta, internal)" + ) + type: str = Field(..., description="The type of the attribute") + value: str = Field(..., description="The value of the attribute") + parentId: Optional[int] = Field(None, description="The ID of the parent object") + autoSync: str = Field(..., description="The auto sync mode") + languageCode: Optional[str] = Field(None, description="The language code of the attribute") + modifiedAt: Optional[str] = Field( + None, description="The date and time the attribute was modified" + ) + modifiedBy: Optional[int] = Field( + None, description="The ID of the user who modified the attribute" + ) + inherited: bool = Field(False, description="Whether the attribute is inherited") + + +class MediaFileResponse(BaseModel): + """Media file metadata""" + + id: int = Field(..., description="The ID of the media file") + clientId: int = Field(..., description="The ID of the client") + mediaId: int = Field(..., description="The ID of the media descriptor") + languageCode: str = Field(..., description="The language code for this media file") + mimeType: str = Field(..., description="The MIME type of the media file") + sourceMimeType: str = Field(..., description="The original MIME type before any conversion") + mamSystem: str = Field(..., description="The Media Asset Management system name") + mamId1: Optional[str] = Field(None, description="MAM system identifier 1") + mamId2: Optional[str] = Field(None, description="MAM system identifier 2") + mamId3: Optional[str] = Field(None, description="MAM system identifier 3") + mamId4: Optional[str] = Field(None, description="MAM system identifier 4") + contentLength: int = Field(..., description="The size of the media file in bytes") + updateCount: int = Field( + ..., description="The number of times this media file has been updated" + ) + changedAt: Optional[str] = Field( + None, description="The date and time the media file was last changed" + ) + changedBy: Optional[int] = Field( + None, description="The ID of the user who last changed the media file" + ) + + +class SingleMediaResponse(BaseModel): + """Complete media descriptor""" + + id: int = Field(..., description="The ID of the media content") + name: str = Field(..., description="The name of the media") + treeId: int = Field(..., description="The ID of the tree this media belongs to") + clientId: int = Field(..., description="The ID of the client") + attributeGroupId: int = Field(..., description="The ID of the media default attribute group") + pictureTypeId: Optional[int] = Field(None, description="The ID of the picture type") + originalId: Optional[int] = Field( + None, description="The ID of the original media if this is a copy" + ) + objectStatus: Optional[str] = Field( + None, description="The status of the object (original, copy)" + ) + userObjectStatus: Optional[int] = Field(None, description="User-defined object status ID") + createdAt: Optional[str] = Field(None, description="The date and time the media was created") + createdBy: Optional[int] = Field(None, description="The ID of the user who created the media") + modifiedAt: Optional[str] = Field( + None, description="The date and time the media was last modified" + ) + modifiedBy: Optional[int] = Field( + None, description="The ID of the user who last modified the media" + ) + files: List[MediaFileResponse] = Field( + default_factory=list, description="The files associated with this media" + ) + attributes: List[AttributeResponse] = Field( + default_factory=list, description="The attribute values of the media" + ) + translations: Optional[Dict[str, str]] = Field( + None, description="Translations of the media name by language code" + ) + + +class SingleNewMediaRequestBody(BaseModel): + """Request body for creating a new media descriptor""" + + name: str = Field(..., description="Name of the media item") + attributeGroupId: Optional[int] = Field( + None, description="The ID of the media default attribute group" + ) + pictureTypeId: Optional[int] = Field(None, description="The ID of the picture type") + treeId: Optional[int] = Field(None, description="The ID of the tree this media belongs to") + originalId: Optional[int] = Field( + None, description="If this is a copy, the ID of the original media" + ) + objectStatus: Optional[str] = Field(None, description="The status of the object") + userObjectStatus: Optional[int] = Field(None, description="Custom user object status ID") + attributes: Optional[List[Dict[str, Any]]] = Field(None, description="List of media attributes") + translations: Optional[Dict[str, str]] = Field( + None, description="Translations of the media name" + ) + + +class SingleUpdateMediaRequestBody(BaseModel): + """Request body for updating a media descriptor""" + + id: int = Field(..., description="The ID of the media descriptor to update") + name: Optional[str] = Field(None, description="Name of the media item") + attributeGroupId: Optional[int] = Field( + None, description="The ID of the media default attribute group" + ) + pictureTypeId: Optional[int] = Field(None, description="The ID of the picture type") + treeId: Optional[int] = Field(None, description="The ID of the tree this media belongs to") + originalId: Optional[int] = Field( + None, description="If this is a copy, the ID of the original media" + ) + objectStatus: Optional[str] = Field(None, description="The status of the object") + userObjectStatus: Optional[int] = Field(None, description="Custom user object status ID") + attributes: Optional[List[Dict[str, Any]]] = Field(None, description="List of media attributes") + translations: Optional[Dict[str, str]] = Field( + None, description="Translations of the media name" + ) + + +class MediaListResponse(BaseModel): + """Paginated list of media descriptors""" + + items: List[SingleMediaResponse] = Field(..., description="List of media items") + total: int = Field(..., description="The total number of media items") + page: int = Field(..., description="The current page number") + limit: int = Field(..., description="The number of media items per page") + links: PaginationLinks = Field(..., description="Pagination links") + + +class MediaBulkCreateResponse(BaseModel): + """Response from bulk media creation""" + + items: List[SingleMediaResponse] = Field(..., description="List of created media items") + totalItemsCreated: int = Field(..., description="The total number of media items created") + + +class MediaBulkUpdateResponse(BaseModel): + """Response from bulk media update""" + + items: List[SingleMediaResponse] = Field(..., description="List of updated media items") + totalItemsUpdated: int = Field(..., description="The total number of media items updated") + + +# ============= Hierarchy Models ============= + + +class HierarchyNode(BaseModel): + """A node in a hierarchy tree""" + + id: int = Field(..., description="The ID of the node") + name: str = Field(..., description="The name of the node") + type: str = Field(..., description="The type of node (product, variant, media, text, etc.)") + children: List["HierarchyNode"] = Field( + default_factory=list, description="The immediate children of the node" + ) + + +HierarchyNode.model_rebuild() + + +class ProductHierarchyResponse(HierarchyNode): + """Hierarchical product structure""" + + pass + + +class ProductGroupHierarchyResponse(BaseModel): + """Hierarchical product group structure""" + + id: int = Field(..., description="The ID of the node") + name: str = Field(..., description="The name of the node") + type: str = Field( + ..., description="The type of node (product-group, product, variant, media, text)" + ) + children: List["ProductGroupHierarchyResponse"] = Field( + default_factory=list, description="The immediate children" + ) + + +ProductGroupHierarchyResponse.model_rebuild() + + +class TreeGroupHierarchyResponse(BaseModel): + """Hierarchical tree group structure""" + + id: int = Field(..., description="The ID of the node") + name: str = Field(..., description="The name of the node") + type: str = Field(..., description="The type of node (tree-group, product-group)") + children: List["TreeGroupHierarchyResponse"] = Field( + default_factory=list, description="The immediate children" + ) + + +TreeGroupHierarchyResponse.model_rebuild() + + +class AttributeGroupHierarchyResponse(BaseModel): + """Hierarchical attribute group structure""" + + id: int = Field(..., description="The ID of the node") + name: str = Field(..., description="The name of the node") + type: str = Field(..., description="The type of node") + children: List["AttributeGroupHierarchyResponse"] = Field( + default_factory=list, description="The immediate children" + ) + + +AttributeGroupHierarchyResponse.model_rebuild()