diff --git a/elytra_client/__init__.py b/elytra_client/__init__.py index 6dbfcfa..f3a750c 100644 --- a/elytra_client/__init__.py +++ b/elytra_client/__init__.py @@ -1,6 +1,6 @@ """Elytra PIM Client - A Pythonic client for the Elytra PIM API""" -__version__ = "0.7.0" +__version__ = "0.3.0" __author__ = "Your Name" from . import rest_api @@ -8,47 +8,27 @@ from .client import ElytraClient from .exceptions import ElytraAPIError, ElytraAuthenticationError from .models import ( ProductAttributeResponse, - SingleMediaResponse, - SingleNewMediaRequestBody, SingleNewProductGroupRequestBody, SingleNewProductRequestBody, - SingleNewTextRequestBody, - SingleNewTreeGroupRequestBody, SingleProductGroupResponse, SingleProductResponse, - SingleTextResponse, - SingleTreeGroupResponse, - SingleUpdateMediaRequestBody, SingleUpdateProductGroupRequestBody, SingleUpdateProductRequestBody, - SingleUpdateTextRequestBody, - SingleUpdateTreeGroupRequestBody, ) __all__ = [ "ElytraClient", "ElytraAPIError", "ElytraAuthenticationError", - # Product models + # Response models "SingleProductResponse", "SingleProductGroupResponse", "ProductAttributeResponse", + # Request models "SingleNewProductRequestBody", "SingleUpdateProductRequestBody", "SingleNewProductGroupRequestBody", "SingleUpdateProductGroupRequestBody", - # Media models - "SingleMediaResponse", - "SingleNewMediaRequestBody", - "SingleUpdateMediaRequestBody", - # Text models - "SingleTextResponse", - "SingleNewTextRequestBody", - "SingleUpdateTextRequestBody", - # Tree group models - "SingleTreeGroupResponse", - "SingleNewTreeGroupRequestBody", - "SingleUpdateTreeGroupRequestBody", # Legacy REST API subpackage "rest_api", ] diff --git a/elytra_client/client.py b/elytra_client/client.py index ba25ecc..25d1213 100644 --- a/elytra_client/client.py +++ b/elytra_client/client.py @@ -1,9 +1,8 @@ """Main Elytra PIM API Client - Fully Pydantic Driven""" -from typing import Any, Dict, List, Optional, Type, TypeVar, cast -from urllib.parse import urljoin - import requests +from typing import Optional, Dict, Any, List, TypeVar, Type, cast +from urllib.parse import urljoin from pydantic import BaseModel, ValidationError from .exceptions import ( @@ -13,35 +12,15 @@ from .exceptions import ( ElytraValidationError, ) from .models import ( - AttributeGroupHierarchyResponse, - MediaFileResponse, - ProductGroupHierarchyResponse, - ProductHierarchyResponse, - SingleAttributeGroupResponse, - SingleAttributeResponse, - SingleMediaResponse, - SingleNewAttributeGroupRequestBody, - SingleNewAttributeRequestBody, - SingleNewMediaRequestBody, - SingleNewProductGroupRequestBody, - SingleNewProductRequestBody, - SingleNewTextRequestBody, - SingleNewTreeGroupRequestBody, - SingleProductGroupResponse, SingleProductResponse, - SingleTextResponse, - SingleTreeGroupResponse, - SingleUpdateAttributeGroupRequestBody, - SingleUpdateAttributeRequestBody, - SingleUpdateMediaRequestBody, - SingleUpdateProductGroupRequestBody, + SingleProductGroupResponse, + SingleNewProductRequestBody, SingleUpdateProductRequestBody, - SingleUpdateTextRequestBody, - SingleUpdateTreeGroupRequestBody, - TreeGroupHierarchyResponse, + SingleNewProductGroupRequestBody, + SingleUpdateProductGroupRequestBody, ) -T = TypeVar("T", bound=BaseModel) +T = TypeVar('T', bound=BaseModel) class ElytraClient: @@ -108,7 +87,7 @@ class ElytraClient: ElytraAPIError: For other API errors """ url = urljoin(self.base_url, endpoint) - + # Convert Pydantic model to dict if needed json_payload = None if json_data is not None: @@ -127,7 +106,7 @@ class ElytraClient: ) response.raise_for_status() data = response.json() - + # Validate response with Pydantic model if provided if response_model is not None: try: @@ -136,11 +115,11 @@ class ElytraClient: raise ElytraValidationError( f"Response validation failed: {e}", response.status_code ) - + return data except requests.exceptions.HTTPError as e: self._handle_http_error(e) - raise # Re-raise after handling to satisfy type checker, + raise # Re-raise after handling to satisfy type checker, # even though _handle_http_error will raise a specific exception and this line should never be reached except requests.exceptions.RequestException as e: raise ElytraAPIError(f"Request failed: {str(e)}") @@ -190,14 +169,16 @@ class ElytraClient: params["groupId"] = group_id response = self._make_request("GET", "/products", params=params) - + # Validate items with Pydantic models if isinstance(response, dict) and "items" in response: try: - response["items"] = [SingleProductResponse(**item) for item in response["items"]] + response["items"] = [ + SingleProductResponse(**item) for item in response["items"] + ] except ValidationError as e: raise ElytraValidationError(f"Product list validation failed: {e}") - + return response def get_product(self, product_id: int, lang: str = "en") -> SingleProductResponse: @@ -240,7 +221,7 @@ class ElytraClient: """ params = {"lang": lang, "page": page, "limit": limit} response = self._make_request("GET", "/groups", params=params) - + # Validate items with Pydantic models if isinstance(response, dict) and "items" in response: try: @@ -249,7 +230,7 @@ class ElytraClient: ] except ValidationError as e: raise ElytraValidationError(f"Product group list validation failed: {e}") - + return response def get_product_group(self, group_id: int, lang: str = "en") -> SingleProductGroupResponse: @@ -276,7 +257,9 @@ class ElytraClient: # Attribute endpoints - def get_attributes(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: + def get_attributes( + self, lang: str = "en", page: int = 1, limit: int = 10 + ) -> Dict[str, Any]: """ Get all attributes. @@ -290,7 +273,7 @@ class ElytraClient: """ params = {"lang": lang, "page": page, "limit": limit} response = self._make_request("GET", "/attributes", params=params) - + return response def get_attribute(self, attribute_id: int, lang: str = "en") -> Dict[str, Any]: @@ -309,7 +292,9 @@ class ElytraClient: # Product CRUD operations with Pydantic validation - def create_product(self, product_data: SingleNewProductRequestBody) -> SingleProductResponse: + def create_product( + self, product_data: SingleNewProductRequestBody + ) -> SingleProductResponse: """ Create a new product. @@ -329,7 +314,9 @@ class ElytraClient: ), ) - def update_product(self, product_data: SingleUpdateProductRequestBody) -> SingleProductResponse: + def update_product( + self, product_data: SingleUpdateProductRequestBody + ) -> SingleProductResponse: """ Update an existing product. @@ -419,937 +406,6 @@ class ElytraClient: """ return self._make_request("DELETE", f"/groups/{group_id}") - # Tree Group endpoints - - def get_tree_groups(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: - """ - Get all tree groups. - - Args: - lang: Language code - page: Page number - limit: Number of tree groups per page - - Returns: - Dictionary containing tree groups list (validated Pydantic models) and pagination info - """ - params = {"lang": lang, "page": page, "limit": limit} - response = self._make_request("GET", "/tree/groups", params=params) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleTreeGroupResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Tree group list validation failed: {e}") - - return response - - def get_tree_group(self, tree_group_id: int, lang: str = "en") -> SingleTreeGroupResponse: - """ - Get a single tree group by ID with its children. - - Args: - tree_group_id: The tree group ID - lang: Language code - - Returns: - Tree group details with children (Pydantic model) - """ - params = {"lang": lang} - return cast( - SingleTreeGroupResponse, - self._make_request( - "GET", - f"/tree/groups/{tree_group_id}", - params=params, - response_model=SingleTreeGroupResponse, - ), - ) - - def create_tree_group( - self, tree_group_data: SingleNewTreeGroupRequestBody - ) -> SingleTreeGroupResponse: - """ - Create a new tree group. - - Args: - tree_group_data: Tree group data (Pydantic model) - - Returns: - Created tree group details (Pydantic model) - """ - return cast( - SingleTreeGroupResponse, - self._make_request( - "POST", - "/tree/groups", - json_data=tree_group_data, - response_model=SingleTreeGroupResponse, - ), - ) - - def update_tree_group( - self, tree_group_data: SingleUpdateTreeGroupRequestBody - ) -> SingleTreeGroupResponse: - """ - Update an existing tree group. - - Args: - tree_group_data: Updated tree group data (Pydantic model) - - Returns: - Updated tree group details (Pydantic model) - """ - return cast( - SingleTreeGroupResponse, - self._make_request( - "PATCH", - "/tree/groups", - json_data=tree_group_data, - response_model=SingleTreeGroupResponse, - ), - ) - - def delete_tree_group(self, tree_group_id: int) -> Dict[str, Any]: - """ - Delete a tree group by ID. - - Args: - tree_group_id: The tree group ID - - Returns: - Deletion response - """ - return self._make_request("DELETE", f"/tree/groups/{tree_group_id}") - - def get_tree_group_hierarchy(self, depth: int = 10) -> TreeGroupHierarchyResponse: - """ - Get the hierarchy of tree groups. - - Args: - depth: The depth of the hierarchy (default: 10) - - Returns: - Tree group hierarchy (Pydantic model) - """ - params = {"depth": depth} - return cast( - TreeGroupHierarchyResponse, - self._make_request( - "GET", - "/tree/groups/hierarchy", - params=params, - response_model=TreeGroupHierarchyResponse, - ), - ) - - # Media endpoints - - def get_media(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: - """ - Get all media items. - - Args: - lang: Language code - page: Page number - limit: Number of media items per page - - Returns: - Dictionary containing media list (validated Pydantic models) and pagination info - """ - params = {"lang": lang, "page": page, "limit": limit} - response = self._make_request("GET", "/media", params=params) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleMediaResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Media list validation failed: {e}") - - return response - - def get_media_by_id(self, media_id: int, lang: str = "en") -> SingleMediaResponse: - """ - Get a single media item by ID. - - Args: - media_id: The media ID - lang: Language code - - Returns: - Media details (Pydantic model) - """ - params = {"lang": lang} - return cast( - SingleMediaResponse, - self._make_request( - "GET", - f"/media/{media_id}", - params=params, - response_model=SingleMediaResponse, - ), - ) - - def create_media(self, media_data: SingleNewMediaRequestBody) -> SingleMediaResponse: - """ - Create a new media item. - - Args: - media_data: Media data (Pydantic model) - - Returns: - Created media details (Pydantic model) - """ - return cast( - SingleMediaResponse, - self._make_request( - "POST", - "/media", - json_data=media_data, - response_model=SingleMediaResponse, - ), - ) - - def update_media(self, media_data: SingleUpdateMediaRequestBody) -> SingleMediaResponse: - """ - Update an existing media item. - - Args: - media_data: Updated media data (Pydantic model) - - Returns: - Updated media details (Pydantic model) - """ - return cast( - SingleMediaResponse, - self._make_request( - "PATCH", - "/media", - json_data=media_data, - response_model=SingleMediaResponse, - ), - ) - - def delete_media(self, media_id: int) -> Dict[str, Any]: - """ - Delete a media item by ID. - - Args: - media_id: The media ID - - Returns: - Deletion response - """ - return self._make_request("DELETE", f"/media/{media_id}") - - def upload_media_file( - self, - file_bytes: bytes, - media_id: int, - mam_system: str, - language_code: Optional[str] = None, - filename: str = "upload", - ) -> Dict[str, Any]: - """ - Upload a file for a media item using multipart/form-data. - - Args: - file_bytes: Raw file content as bytes - media_id: The Elytra media ID to attach the file to - mam_system: MAM system identifier (e.g. 'fs', 'sixomc', 'cumulus') - language_code: Optional language code for the file - filename: Filename hint sent with the upload - - Returns: - Upload response dict - """ - url = urljoin(self.base_url, "/media/file") - - form_data: Dict[str, Any] = { - "mediaId": str(media_id), - "mamSystem": mam_system, - } - if language_code: - form_data["languageCode"] = language_code - - # Build request manually so requests can set the multipart Content-Type boundary - # without being overridden by the session-level Content-Type: application/json header. - req = requests.Request( - "POST", - url, - headers={"Authorization": f"Bearer {self.api_key}", "Accept": "application/json"}, - files={"file": (filename, file_bytes)}, - data=form_data, - ) - prepared = req.prepare() - - try: - response = self.session.send(prepared, timeout=self.timeout) - response.raise_for_status() - except requests.exceptions.HTTPError as e: - self._handle_http_error(e) - raise - - return response.json() - - def download_media_file(self, media_file_id: int) -> bytes: - """ - Download a media file by its file ID. - - Args: - media_file_id: The media file ID - - Returns: - File content as raw bytes - """ - url = urljoin(self.base_url, f"/media/file/{media_file_id}") - - try: - response = self.session.get(url, timeout=self.timeout) - response.raise_for_status() - except requests.exceptions.HTTPError as e: - self._handle_http_error(e) - raise - - return response.content - - def delete_media_file(self, media_file_id: int) -> Dict[str, Any]: - """ - Delete a media file. - - Args: - media_file_id: The media file ID - - Returns: - Deletion response - """ - return self._make_request("DELETE", f"/media/file/{media_file_id}") - - # Text endpoints - - def get_texts(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: - """ - Get all text items. - - Args: - lang: Language code - page: Page number - limit: Number of text items per page - - Returns: - Dictionary containing text list (validated Pydantic models) and pagination info - """ - params = {"lang": lang, "page": page, "limit": limit} - response = self._make_request("GET", "/text", params=params) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleTextResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Text list validation failed: {e}") - - return response - - def get_text(self, text_id: int, lang: str = "en") -> SingleTextResponse: - """ - Get a single text item by ID. - - Args: - text_id: The text ID - lang: Language code - - Returns: - Text details (Pydantic model) - """ - params = {"lang": lang} - return cast( - SingleTextResponse, - self._make_request( - "GET", - f"/text/{text_id}", - params=params, - response_model=SingleTextResponse, - ), - ) - - def create_text(self, text_data: SingleNewTextRequestBody) -> SingleTextResponse: - """ - Create a new text item. - - Args: - text_data: Text data (Pydantic model) - - Returns: - Created text details (Pydantic model) - """ - return cast( - SingleTextResponse, - self._make_request( - "POST", - "/text", - json_data=text_data, - response_model=SingleTextResponse, - ), - ) - - def update_text(self, text_data: SingleUpdateTextRequestBody) -> SingleTextResponse: - """ - Update an existing text item. - - Args: - text_data: Updated text data (Pydantic model) - - Returns: - Updated text details (Pydantic model) - """ - return cast( - SingleTextResponse, - self._make_request( - "PATCH", - "/text", - json_data=text_data, - response_model=SingleTextResponse, - ), - ) - - def delete_text(self, text_id: int) -> Dict[str, Any]: - """ - Delete a text item by ID. - - Args: - text_id: The text ID - - Returns: - Deletion response - """ - return self._make_request("DELETE", f"/text/{text_id}") - - # Attribute extended endpoints - - def create_attribute( - self, attribute_data: SingleNewAttributeRequestBody - ) -> SingleAttributeResponse: - """ - Create a new attribute definition. - - Args: - attribute_data: Attribute definition data (Pydantic model) - - Returns: - Created attribute details (Pydantic model) - """ - return cast( - SingleAttributeResponse, - self._make_request( - "POST", - "/attributes", - json_data=attribute_data, - response_model=SingleAttributeResponse, - ), - ) - - def update_attribute( - self, attribute_data: SingleUpdateAttributeRequestBody - ) -> SingleAttributeResponse: - """ - Update an existing attribute definition. - - Args: - attribute_data: Updated attribute definition data (Pydantic model) - - Returns: - Updated attribute details (Pydantic model) - """ - return cast( - SingleAttributeResponse, - self._make_request( - "PATCH", - "/attributes", - json_data=attribute_data, - response_model=SingleAttributeResponse, - ), - ) - - def delete_attribute(self, attribute_id: int) -> Dict[str, Any]: - """ - Delete an attribute definition by ID. - - Args: - attribute_id: The attribute ID - - Returns: - Deletion response - """ - return self._make_request("DELETE", f"/attributes/{attribute_id}") - - def get_attribute_by_name( - self, attribute_name: str, lang: str = "en" - ) -> SingleAttributeResponse: - """ - Get an attribute definition by name. - - Args: - attribute_name: The attribute name (independent name) - lang: Language code - - Returns: - Attribute details (Pydantic model) - """ - params = {"lang": lang} - return cast( - SingleAttributeResponse, - self._make_request( - "GET", - f"/attributes/name/{attribute_name}", - params=params, - response_model=SingleAttributeResponse, - ), - ) - - # Attribute Group endpoints - - def get_attribute_groups( - self, lang: str = "en", page: int = 1, limit: int = 10 - ) -> Dict[str, Any]: - """ - Get all attribute groups. - - Args: - lang: Language code - page: Page number - limit: Number of attribute groups per page - - Returns: - Dictionary containing attribute groups list (validated Pydantic models) and pagination info - """ - params = {"lang": lang, "page": page, "limit": limit} - response = self._make_request("GET", "/attribute/groups", params=params) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [ - SingleAttributeGroupResponse(**item) for item in response["items"] - ] - except ValidationError as e: - raise ElytraValidationError(f"Attribute group list validation failed: {e}") - - return response - - def create_attribute_group( - self, attribute_group_data: SingleNewAttributeGroupRequestBody - ) -> SingleAttributeGroupResponse: - """ - Create a new attribute group. - - Args: - attribute_group_data: Attribute group data (Pydantic model) - - Returns: - Created attribute group details (Pydantic model) - """ - return cast( - SingleAttributeGroupResponse, - self._make_request( - "POST", - "/attribute/groups", - json_data=attribute_group_data, - response_model=SingleAttributeGroupResponse, - ), - ) - - def update_attribute_group( - self, attribute_group_data: SingleUpdateAttributeGroupRequestBody - ) -> SingleAttributeGroupResponse: - """ - Update an existing attribute group. - - Args: - attribute_group_data: Updated attribute group data (Pydantic model) - - Returns: - Updated attribute group details (Pydantic model) - """ - return cast( - SingleAttributeGroupResponse, - self._make_request( - "PATCH", - "/attribute/groups", - json_data=attribute_group_data, - response_model=SingleAttributeGroupResponse, - ), - ) - - def get_attribute_group_by_id( - self, attribute_group_id: int, lang: str = "en" - ) -> SingleAttributeGroupResponse: - """ - Get an attribute group by ID. - - Args: - attribute_group_id: The attribute group ID - lang: Language code - - Returns: - Attribute group details (Pydantic model) - """ - params = {"lang": lang} - return cast( - SingleAttributeGroupResponse, - self._make_request( - "GET", - f"/attribute/groups/id/{attribute_group_id}", - params=params, - response_model=SingleAttributeGroupResponse, - ), - ) - - def get_attribute_group_by_name( - self, attribute_group_name: str, lang: str = "en" - ) -> SingleAttributeGroupResponse: - """ - Get an attribute group by name. - - Args: - attribute_group_name: The attribute group name - lang: Language code - - Returns: - Attribute group details (Pydantic model) - """ - params = {"lang": lang} - return cast( - SingleAttributeGroupResponse, - self._make_request( - "GET", - f"/attribute/groups/name/{attribute_group_name}", - params=params, - response_model=SingleAttributeGroupResponse, - ), - ) - - def add_attributes_to_group( - self, attribute_group_id: int, attribute_ids: List[int] - ) -> Dict[str, Any]: - """ - Add attributes to an attribute group. - - Args: - attribute_group_id: The attribute group ID - attribute_ids: List of attribute IDs to add - - Returns: - Operation response - """ - data = { - "attributeGroupId": attribute_group_id, - "attributeIds": attribute_ids, - } - return self._make_request( - "POST", - "/attribute/groups/operations/add", - json_data=data, - ) - - def get_attribute_group_hierarchy( - self, attribute_group_id: int - ) -> AttributeGroupHierarchyResponse: - """ - Get the hierarchy of an attribute group. - - Args: - attribute_group_id: The attribute group ID - - Returns: - Attribute group hierarchy (Pydantic model) - """ - return cast( - AttributeGroupHierarchyResponse, - self._make_request( - "GET", - f"/attribute/groups/hierarchy/{attribute_group_id}", - response_model=AttributeGroupHierarchyResponse, - ), - ) - - # Hierarchy endpoints - - def get_product_hierarchy(self, product_id: int, depth: int = 10) -> ProductHierarchyResponse: - """ - Get the hierarchy of a product. - - Args: - product_id: The product ID - depth: The depth of the hierarchy (default: 10) - - Returns: - Product hierarchy (Pydantic model) - """ - params = {"depth": depth} - return cast( - ProductHierarchyResponse, - self._make_request( - "GET", - f"/products/{product_id}/hierarchy", - params=params, - response_model=ProductHierarchyResponse, - ), - ) - - def get_product_group_hierarchy( - self, group_id: int, depth: int = 10 - ) -> ProductGroupHierarchyResponse: - """ - Get the hierarchy of a product group. - - Args: - group_id: The product group ID - depth: The depth of the hierarchy (default: 10) - - Returns: - Product group hierarchy (Pydantic model) - """ - params = {"depth": depth} - return cast( - ProductGroupHierarchyResponse, - self._make_request( - "GET", - f"/groups/{group_id}/hierarchy", - params=params, - response_model=ProductGroupHierarchyResponse, - ), - ) - - # Product operations - - def perform_product_operation( - self, - operation: str, - product_id: int, - parent_id: int, - ) -> SingleProductResponse: - """ - Perform an operation on a product. - - Available operations: - - **copy**: Copy the product with all children to a new parent. Requires a product group as parent. - - **move**: Move the product to a new parent. Accepts product or product group as parent. - - **link**: Link the product to a new parent. Accepts product or product group as parent. - - **copy-structure**: Copy the group structure but link products. Requires a product group as parent. - - Args: - operation: The operation to perform (copy, move, link, copy-structure) - product_id: The ID of the product to perform the operation on - parent_id: The ID of the destination parent - - Returns: - Updated product details (Pydantic model) - - Raises: - ElytraValidationError: If operation requires specific parent type not provided - """ - data = { - "operation": operation, - "productId": product_id, - "parentId": parent_id, - } - return cast( - SingleProductResponse, - self._make_request( - "POST", - "/products/operation", - json_data=data, - response_model=SingleProductResponse, - ), - ) - - # Bulk operations - Products - - def create_products_bulk( - self, products_data: List[SingleNewProductRequestBody] - ) -> Dict[str, Any]: - """ - Create multiple products in bulk. - - Args: - products_data: List of product data (Pydantic models) - - Returns: - Response with created items and total count - """ - response = self._make_request("POST", "/products/bulk", json_data=products_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleProductResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk products creation validation failed: {e}") - - return response - - def update_products_bulk( - self, products_data: List[SingleUpdateProductRequestBody] - ) -> Dict[str, Any]: - """ - Update multiple products in bulk. - - Args: - products_data: List of updated product data (Pydantic models) - - Returns: - Response with updated items and total count - """ - response = self._make_request("PATCH", "/products/bulk", json_data=products_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleProductResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk products update validation failed: {e}") - - return response - - # Bulk operations - Attributes - - def create_attributes_bulk( - self, attributes_data: List[SingleNewAttributeRequestBody] - ) -> Dict[str, Any]: - """ - Create multiple attribute definitions in bulk. - - Args: - attributes_data: List of attribute definition data (Pydantic models) - - Returns: - Response with created items and total count - """ - response = self._make_request("POST", "/attributes/bulk", json_data=attributes_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleAttributeResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk attributes creation validation failed: {e}") - - return response - - def update_attributes_bulk( - self, attributes_data: List[SingleUpdateAttributeRequestBody] - ) -> Dict[str, Any]: - """ - Update multiple attribute definitions in bulk. - - Args: - attributes_data: List of updated attribute definition data (Pydantic models) - - Returns: - Response with updated items and total count - """ - response = self._make_request("PATCH", "/attributes/bulk", json_data=attributes_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleAttributeResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk attributes update validation failed: {e}") - - return response - - # Bulk operations - Media - - def create_media_bulk(self, media_data: List[SingleNewMediaRequestBody]) -> Dict[str, Any]: - """ - Create multiple media items in bulk. - - Args: - media_data: List of media data (Pydantic models) - - Returns: - Response with created items and total count - """ - response = self._make_request("POST", "/media/bulk", json_data=media_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleMediaResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk media creation validation failed: {e}") - - return response - - def update_media_bulk(self, media_data: List[SingleUpdateMediaRequestBody]) -> Dict[str, Any]: - """ - Update multiple media items in bulk. - - Args: - media_data: List of updated media data (Pydantic models) - - Returns: - Response with updated items and total count - """ - response = self._make_request("PATCH", "/media/bulk", json_data=media_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleMediaResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk media update validation failed: {e}") - - return response - - # Bulk operations - Text - - def create_texts_bulk(self, texts_data: List[SingleNewTextRequestBody]) -> Dict[str, Any]: - """ - Create multiple text items in bulk. - - Args: - texts_data: List of text data (Pydantic models) - - Returns: - Response with created items and total count - """ - response = self._make_request("POST", "/text/bulk", json_data=texts_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleTextResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk text creation validation failed: {e}") - - return response - - def update_texts_bulk(self, texts_data: List[SingleUpdateTextRequestBody]) -> Dict[str, Any]: - """ - Update multiple text items in bulk. - - Args: - texts_data: List of updated text data (Pydantic models) - - Returns: - Response with updated items and total count - """ - response = self._make_request("PATCH", "/text/bulk", json_data=texts_data) - - # Validate items with Pydantic models - if isinstance(response, dict) and "items" in response: - try: - response["items"] = [SingleTextResponse(**item) for item in response["items"]] - except ValidationError as e: - raise ElytraValidationError(f"Bulk text update validation failed: {e}") - - return response - # Health check def health_check(self) -> Dict[str, Any]: @@ -1372,4 +428,3 @@ class ElytraClient: def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit""" self.close() - self.close() diff --git a/elytra_client/rest_api/README.md b/elytra_client/rest_api/README.md index 12c2fbf..215d6ba 100644 --- a/elytra_client/rest_api/README.md +++ b/elytra_client/rest_api/README.md @@ -1,19 +1,11 @@ # Lobster PIM Legacy REST API Client -This subpackage provides a Python client for accessing the **legacy REST API** of Lobster PIM (now called Elytra). - -## ⚠️ Important: Legacy API vs. New Web API - -The Lobster REST API is the **legacy API** that provides read-only access to: -- **Scheduled Jobs** (`/rest/job/*`) - Job execution and monitoring -- **Protocol Logs** (`/rest/protocol/*`) - Execution logs and protocol information - -**For CRUD operations** on products, product groups, attributes, media, and other resources in the new Elytra PIM Web API, use the [`ElytraClient`](../client.py) with the OpenAPI-based Web API instead. +This subpackage provides a Python client for accessing the legacy REST API of Lobster PIM (Product Information Management system). It offers access to scheduled jobs and protocol logs through a clean, Pydantic-based interface. ## Features -- **Job Management**: Get, monitor, and execute scheduled jobs (read-only) -- **Protocol/Log Access**: Retrieve execution logs and protocol information (read-only) +- **Job Management**: Access, monitor, and execute scheduled jobs +- **Protocol/Log Access**: Retrieve execution logs and protocol information - **Authentication**: Support for both username/password and API token authentication - **Job Control**: Execute jobs with parameter overrides and queue management - **Type Safety**: Full Pydantic model validation for all API responses diff --git a/elytra_client/rest_api/__init__.py b/elytra_client/rest_api/__init__.py index bbbcf1f..93f4111 100644 --- a/elytra_client/rest_api/__init__.py +++ b/elytra_client/rest_api/__init__.py @@ -1,11 +1,4 @@ -"""Lobster PIM Legacy REST API client and utilities - -This module provides access to the legacy Lobster REST API, which offers -read-only access to Scheduled Jobs and Protocol logs. - -For CRUD operations on products, product groups, attributes, media, and other -resources in the new Elytra PIM Web API, use the ElytraClient instead. -""" +"""Lobster PIM Legacy REST API client and utilities""" from .auth import AuthMethod, RestApiAuth from .client import LobsterRestApiClient @@ -24,23 +17,18 @@ from .models import ( ) __all__ = [ - # Authentication "RestApiAuth", "AuthMethod", - # Client "LobsterRestApiClient", - # Job models "JobInfo", "JobDetailInfo", "JobOverviewResponse", "JobExecutionResponse", "JobControlRequest", "JobControlResponse", - # Protocol models "ProtocolInfo", "ProtocolListResponse", "ProtocolCategoryInfo", "ProtocolCategoryListResponse", - # Error model "ErrorResponse", ] diff --git a/elytra_client/rest_api/client.py b/elytra_client/rest_api/client.py new file mode 100644 index 0000000..f24b8a2 --- /dev/null +++ b/elytra_client/rest_api/client.py @@ -0,0 +1,438 @@ +"""Client for the Lobster PIM Legacy REST API""" + +from typing import Any, Dict, List, Optional, Type, TypeVar, cast +from urllib.parse import urljoin + +import requests +from pydantic import BaseModel, ValidationError + +from ..exceptions import ( + ElytraAPIError, + ElytraAuthenticationError, + ElytraNotFoundError, + ElytraValidationError, +) +from .auth import AuthMethod, RestApiAuth +from .models import ( + JobControlRequest, + JobControlResponse, + JobDetailInfo, + JobExecutionResponse, + JobInfo, + JobOverviewResponse, + ProtocolCategoryInfo, + ProtocolCategoryListResponse, + ProtocolInfo, + ProtocolListResponse, +) + +T = TypeVar('T', bound=BaseModel) + + +class LobsterRestApiClient: + """ + Client for the Lobster PIM Legacy REST API. + + Provides access to scheduled jobs and protocol logs via REST endpoints. + Supports both username/password and API token authentication. + + Args: + base_url: The base URL of the Lobster PIM server (e.g., http://localhost:8080) + auth: RestApiAuth instance for authentication + timeout: Request timeout in seconds (default: 30) + """ + + def __init__( + self, + base_url: str, + auth: RestApiAuth, + timeout: int = 30, + ): + """Initialize the Lobster REST API client""" + self.base_url = base_url.rstrip("/") + self.auth = auth + self.timeout = timeout + self.session = requests.Session() + self._setup_headers() + + def _setup_headers(self) -> None: + """Set up request headers including authentication""" + self.session.headers.update( + { + "Content-Type": "application/json", + "Accept": "application/json", + } + ) + self.session.headers.update(self.auth.get_auth_header()) + + def _handle_response( + self, + response: requests.Response, + expected_model: Type[T], + ) -> T: + """ + Handle API response and parse into Pydantic model. + + Args: + response: Response from requests + expected_model: Pydantic model to deserialize into + + Returns: + Parsed response as Pydantic model + + Raises: + ElytraAuthenticationError: If authentication fails + ElytraNotFoundError: If resource not found + ElytraAPIError: For other API errors + ElytraValidationError: If response validation fails + """ + if response.status_code == 401: + raise ElytraAuthenticationError("Authentication failed") + elif response.status_code == 404: + raise ElytraNotFoundError("Resource not found") + elif response.status_code == 429: + raise ElytraAPIError("Too many requests - rate limit exceeded") + elif response.status_code >= 400: + try: + error_data = response.json() + error_msg = error_data.get("error") or error_data.get("message", response.text) + except Exception: + error_msg = response.text + raise ElytraAPIError(f"API error {response.status_code}: {error_msg}") + + try: + data = response.json() + except Exception as e: + raise ElytraAPIError(f"Failed to parse response as JSON: {str(e)}") + + try: + return expected_model.model_validate(data) + except ValidationError as e: + raise ElytraValidationError(f"Response validation failed: {str(e)}") + + def _make_request( + self, + method: str, + endpoint: str, + expected_model: Type[T], + params: Optional[Dict[str, Any]] = None, + json_data: Optional[Dict[str, Any]] = None, + ) -> T: + """ + Make HTTP request to the REST API. + + Args: + method: HTTP method (GET, POST, etc.) + endpoint: API endpoint path + expected_model: Pydantic model for response + params: Query parameters + json_data: JSON request body + + Returns: + Parsed response as Pydantic model + """ + url = urljoin(self.base_url, f"/rest/{endpoint}") + + # Add authentication parameters for GET requests + if method.upper() == "GET" and self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + if params is None: + params = {} + params.update(self.auth.get_url_parameters()) + + response = self.session.request( + method=method, + url=url, + params=params, + json=json_data, + timeout=self.timeout, + ) + + return self._handle_response(response, expected_model) + + # ============= Job Endpoints ============= + + def get_job_overview(self) -> JobOverviewResponse: + """ + Get overview of all active jobs. + + Returns: + JobOverviewResponse containing list of active jobs + """ + return self._make_request( + "GET", + "job/overview", + JobOverviewResponse, + ) + + def get_job_html_overview(self) -> str: + """ + Get HTML overview of all active jobs. + + Returns: + HTML page content with job overview + """ + url = urljoin(self.base_url, "/rest/job/overview") + 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 get job overview: {response.status_code}") + return response.text + + def get_all_active_jobs(self) -> JobOverviewResponse: + """ + Get list of all active jobs. + + Returns: + JobOverviewResponse containing list of all active jobs + """ + return self._make_request( + "GET", + "job/job_id", + JobOverviewResponse, + ) + + def get_job_detail(self, job_id: int) -> JobDetailInfo: + """ + Get details of a specific active job. + + Args: + job_id: ID of the job + + Returns: + JobDetailInfo with job details + """ + return self._make_request( + "GET", + f"job/job_id/{job_id}", + JobDetailInfo, + ) + + def execute_job( + self, + job_id: int, + ) -> JobExecutionResponse: + """ + Execute a job and get details of the started job. + + Args: + job_id: ID of the job to execute + + Returns: + JobExecutionResponse with execution details + """ + return self._make_request( + "GET", + f"job/execute/{job_id}", + JobExecutionResponse, + ) + + def get_running_job_instances(self) -> JobOverviewResponse: + """ + Get list of running job instances. + + Returns: + JobOverviewResponse containing list of running job instances + """ + return self._make_request( + "GET", + "job/runtime_id", + JobOverviewResponse, + ) + + def get_running_job_instance(self, runtime_id: str) -> JobOverviewResponse: + """ + Get details of a specific running job instance. + + Args: + runtime_id: Runtime ID of the job instance + + Returns: + JobOverviewResponse with instance details + """ + return self._make_request( + "GET", + f"job/runtime_id/{runtime_id}", + JobOverviewResponse, + ) + + def control_job( + self, + job_id: int, + action: str = "start", + additional_reference: Optional[str] = None, + parameters: Optional[Dict[str, Any]] = None, + queue_id: Optional[str] = None, + max_job_duration_seconds: Optional[int] = None, + ) -> JobControlResponse: + """ + Control a job using the control endpoint (POST). + + Supports starting jobs with parameter overrides and queue management. + + Args: + job_id: ID of the job to control + action: Action to perform (default: "start") + additional_reference: Optional reference for external tracking + parameters: Optional parameters to override job settings + queue_id: Optional queue ID for serialized execution + max_job_duration_seconds: Max duration in seconds (default 12 hours) + + Returns: + JobControlResponse with execution details + """ + request_body = { + "action": action, + "objectId": job_id, + "objectType": "job", + } + + # Add authentication credentials for POST + request_body.update(self.auth.get_json_body_params()) + + if additional_reference: + request_body["additionalReference"] = additional_reference + + if parameters: + request_body["parameter"] = parameters + + if queue_id: + request_body["queueId"] = queue_id + + if max_job_duration_seconds: + request_body["maxJobDurationSeconds"] = max_job_duration_seconds + + return self._make_request( + "POST", + "job/control", + JobControlResponse, + json_data=request_body, + ) + + # ============= Protocol Endpoints ============= + + def get_protocols(self, limit: int = 50) -> ProtocolListResponse: + """ + Get the last N protocols. + + Args: + limit: Number of protocols to retrieve (default: 50) + + Returns: + ProtocolListResponse containing list of protocols + """ + response = self.session.get( + urljoin(self.base_url, "/rest/protocol"), + params={"limit": limit, **self.auth.get_url_parameters()}, + timeout=self.timeout, + ) + return self._handle_response(response, ProtocolListResponse) + + def get_protocol(self, protocol_id: str) -> ProtocolInfo: + """ + Get details of a specific protocol. + + Args: + protocol_id: ID of the protocol + + Returns: + ProtocolInfo with protocol details + """ + return self._make_request( + "GET", + f"protocol/{protocol_id}", + ProtocolInfo, + ) + + def get_protocol_by_job_id(self, job_id: int) -> ProtocolListResponse: + """ + Get all protocols for a specific job ID. + + Args: + job_id: ID of the job + + Returns: + ProtocolListResponse containing protocols for the job + """ + return self._make_request( + "GET", + f"protocol/job/{job_id}", + ProtocolListResponse, + ) + + def get_protocol_by_runtime_id(self, runtime_id: str) -> ProtocolInfo: + """ + Get protocol for a specific job instance runtime ID. + + Args: + runtime_id: Runtime ID of the job instance + + Returns: + ProtocolInfo with protocol details + """ + return self._make_request( + "GET", + f"protocol/job/{runtime_id}", + ProtocolInfo, + ) + + def get_protocol_by_additional_reference(self, reference: str) -> ProtocolListResponse: + """ + Get all protocols for a specific additional reference. + + Args: + reference: Additional reference value + + Returns: + ProtocolListResponse containing protocols for the reference + """ + return self._make_request( + "GET", + f"protocol/job/{reference}", + ProtocolListResponse, + ) + + def get_all_protocol_categories(self) -> ProtocolCategoryListResponse: + """ + Get all available protocol categories. + + Returns: + ProtocolCategoryListResponse containing list of categories + """ + return self._make_request( + "GET", + "protocol/category", + ProtocolCategoryListResponse, + ) + + def get_protocol_by_category(self, category_id: str, limit: int = 50) -> ProtocolListResponse: + """ + Get the last N protocols from a specific category. + + Args: + category_id: ID of the protocol category + limit: Number of protocols to retrieve (default: 50) + + Returns: + ProtocolListResponse containing protocols for the category + """ + return self._make_request( + "GET", + f"protocol/category/{category_id}", + ProtocolListResponse, + params={"limit": limit}, + ) + + def close(self) -> None: + """Close the session and clean up resources.""" + self.session.close() + + def __enter__(self): + """Context manager entry""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit""" + self.close() diff --git a/elytra_client/rest_api/client/__init__.py b/elytra_client/rest_api/client/__init__.py deleted file mode 100644 index c33ab6b..0000000 --- a/elytra_client/rest_api/client/__init__.py +++ /dev/null @@ -1,43 +0,0 @@ -"""Lobster PIM Legacy REST API Client.""" - -from .base import LobsterRestApiClientBase -from .jobs import JobsMixin -from .protocols import ProtocolsMixin - - -class LobsterRestApiClient( - LobsterRestApiClientBase, - JobsMixin, - ProtocolsMixin, -): - """ - Legacy REST API client for the Lobster PIM system. - - Provides read-only access to Scheduled Jobs and Protocol logs via the - Lobster REST API endpoints. Only supports GET operations on: - - Scheduled Jobs (/rest/job/*) - - Protocols (/rest/protocol/*) - - The new Elytra PIM Web API (for CRUD operations on products, groups, - attributes, media, etc.) should use the ElytraClient instead. - - Example: - >>> from elytra_client.rest_api.client import LobsterRestApiClient - >>> from elytra_client.rest_api.auth import RestApiAuth, AuthMethod - >>> - >>> auth = RestApiAuth.from_bearer_token("your-token") - >>> client = LobsterRestApiClient("http://localhost:8080", auth) - >>> - >>> # Access jobs (read-only) - >>> jobs = client.get_all_active_jobs() - >>> job_detail = client.get_job_detail(job_id=172475107) - >>> - >>> # Access protocols (read-only) - >>> protocols = client.get_protocols() - >>> protocol = client.get_protocol(protocol_id="176728573") - """ - - pass - - -__all__ = ["LobsterRestApiClient"] diff --git a/elytra_client/rest_api/client/_base_protocol.py b/elytra_client/rest_api/client/_base_protocol.py deleted file mode 100644 index b3efe49..0000000 --- a/elytra_client/rest_api/client/_base_protocol.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Protocol defining the interface expected by all mixin classes.""" - -from typing import Any, Dict, Optional, Protocol, Type, TypeVar - -import requests -from pydantic import BaseModel - -T = TypeVar("T", bound=BaseModel) - - -class ClientBaseProtocol(Protocol): - """Protocol that defines the base client interface for mixins.""" - - base_url: str - auth: Any - timeout: int - session: requests.Session - - def _make_request( - self, - method: str, - endpoint: str, - expected_model: Type[T], - params: Optional[Dict[str, Any]] = None, - json_data: Optional[Dict[str, Any]] = None, - ) -> T: - """Make HTTP request to the REST API.""" - ... - - def _handle_response( - self, - response: requests.Response, - expected_model: Type[T], - ) -> T: - """Handle API response and parse into Pydantic model.""" - ... diff --git a/elytra_client/rest_api/client/attribute_groups.py b/elytra_client/rest_api/client/attribute_groups.py deleted file mode 100644 index 1f4b8cd..0000000 --- a/elytra_client/rest_api/client/attribute_groups.py +++ /dev/null @@ -1,198 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Attribute Groups mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional -from urllib.parse import urljoin - -from ...exceptions import ElytraAPIError -from ..auth import AuthMethod -from ..models import ( - AttributeGroupListResponse, - SingleAttributeGroupResponse, - SingleUpdateAttributeGroupRequestBody, -) - - -class AttributeGroupsMixin: - """Mixin for attribute group-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_all_attribute_groups( - self, - lang: Optional[str] = None, - page: int = 1, - limit: int = 10, - ) -> AttributeGroupListResponse: - """ - Get all attribute groups. - - Args: - lang: Language code (optional) - page: Page number (default: 1) - limit: Number of groups per page (default: 10) - - Returns: - AttributeGroupListResponse with paginated list of attribute groups - """ - params: Dict[str, Any] = {"page": page, "limit": limit} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - "attribute/groups", - AttributeGroupListResponse, - params=params, - ) - - def create_attribute_group(self, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse: - """ - Create a new attribute group. - - Args: - group_data: Attribute group data - - Returns: - SingleAttributeGroupResponse with created group - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "attribute/groups", - SingleAttributeGroupResponse, - json_data=group_data, - ) - - def get_attribute_group_by_id( - self, group_id: int, lang: Optional[str] = None - ) -> SingleAttributeGroupResponse: - """ - Get an attribute group by ID. - - Args: - group_id: ID of the attribute group - lang: Language code (optional) - - Returns: - SingleAttributeGroupResponse with group details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"attribute/groups/{group_id}", - SingleAttributeGroupResponse, - params=params if params else None, - ) - - def delete_attribute_group(self, group_id: int) -> None: - """ - Delete an attribute group by ID. - - Args: - group_id: ID of the attribute group to delete - """ - url = urljoin(self.base_url, f"/rest/attribute/groups/{group_id}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") - - def get_attribute_group_by_name( - self, group_name: str, lang: Optional[str] = None - ) -> SingleAttributeGroupResponse: - """ - Get an attribute group by name. - - Args: - group_name: Name of the attribute group - lang: Language code (optional) - - Returns: - SingleAttributeGroupResponse with group details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"attribute/groups/name/{group_name}", - SingleAttributeGroupResponse, - params=params if params else None, - ) - - def update_attribute_group_by_name( - self, group_name: str, group_data: Dict[str, Any] - ) -> SingleAttributeGroupResponse: - """ - Update an attribute group by name. - - Args: - group_name: Name of the attribute group to update - group_data: Updated group data - - Returns: - SingleAttributeGroupResponse with updated group - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - f"attribute/groups/name/{group_name}", - SingleAttributeGroupResponse, - json_data=group_data, - ) - - def delete_attribute_group_by_name(self, group_name: str) -> None: - """ - Delete an attribute group by name. - - Args: - group_name: Name of the attribute group to delete - """ - url = urljoin( - self.base_url, f"/rest/attribute/groups/name/{group_name}" # type: ignore[attr-defined] - ) - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError( - f"Failed to delete attribute group by name: {response.status_code}" - ) - - def attribute_group_add_operation( - self, operation_data: Dict[str, Any] - ) -> AttributeGroupListResponse: - """ - Perform bulk operations on attribute groups. - - Args: - operation_data: Operation details - - Returns: - AttributeGroupListResponse with affected groups - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "attribute/groups/operation", - AttributeGroupListResponse, - json_data=operation_data, - ) diff --git a/elytra_client/rest_api/client/attributes.py b/elytra_client/rest_api/client/attributes.py deleted file mode 100644 index d8f34af..0000000 --- a/elytra_client/rest_api/client/attributes.py +++ /dev/null @@ -1,220 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Attributes mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional -from urllib.parse import urljoin - -from ...exceptions import ElytraAPIError -from ..auth import AuthMethod -from ..models import ( - AttributeBulkCreateResponse, - AttributeBulkUpdateResponse, - AttributeGetByNameResponse, - AttributeGroupHierarchyResponse, - AttributeListResponse, - SimpleAttributeResponse, -) - - -class AttributesMixin: - """Mixin for attribute-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_all_attributes( - self, - lang: Optional[str] = None, - page: int = 1, - limit: int = 10, - ) -> AttributeListResponse: - """ - Get all attributes. - - Args: - lang: Language code (optional) - page: Page number (default: 1) - limit: Number of attributes per page (default: 10) - - Returns: - AttributeListResponse with paginated list of attributes - """ - params: Dict[str, Any] = {"page": page, "limit": limit} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - "attributes", - AttributeListResponse, - params=params, - ) - - def create_attribute(self, attribute_data: Dict[str, Any]) -> SimpleAttributeResponse: - """ - Create a new attribute. - - Args: - attribute_data: Attribute data - - Returns: - SimpleAttributeResponse with created attribute - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "attributes", - SimpleAttributeResponse, - json_data=attribute_data, - ) - - def update_attribute(self, attribute_data: Dict[str, Any]) -> SimpleAttributeResponse: - """ - Update an attribute. - - Args: - attribute_data: Updated attribute data (must include 'id') - - Returns: - SimpleAttributeResponse with updated attribute - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "attributes", - SimpleAttributeResponse, - json_data=attribute_data, - ) - - def create_multiple_attributes( - self, attributes_list: List[Dict[str, Any]] - ) -> AttributeBulkCreateResponse: - """ - Create multiple attributes in bulk. - - Args: - attributes_list: List of attribute data - - Returns: - AttributeBulkCreateResponse with created attributes - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "attributes/bulk", - AttributeBulkCreateResponse, - json_data=attributes_list, - ) - - def update_multiple_attributes( - self, attributes_list: List[Dict[str, Any]] - ) -> AttributeBulkUpdateResponse: - """ - Update multiple attributes in bulk. - - Args: - attributes_list: List of attribute data to update (each must include 'id') - - Returns: - AttributeBulkUpdateResponse with updated attributes - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "attributes/bulk", - AttributeBulkUpdateResponse, - json_data=attributes_list, - ) - - def get_attribute_by_id( - self, attribute_id: int, lang: Optional[str] = None - ) -> SimpleAttributeResponse: - """ - Get an attribute by ID. - - Args: - attribute_id: ID of the attribute - lang: Language code (optional) - - Returns: - SimpleAttributeResponse with attribute details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"attributes/{attribute_id}", - SimpleAttributeResponse, - params=params if params else None, - ) - - def delete_attribute(self, attribute_id: int) -> None: - """ - Delete an attribute by ID. - - Args: - attribute_id: ID of the attribute to delete - """ - url = urljoin( - self.base_url, f"/rest/attributes/{attribute_id}" # type: ignore[attr-defined] - ) - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete attribute: {response.status_code}") - - def get_attribute_by_name( - self, attribute_name: str, lang: Optional[str] = None - ) -> AttributeGetByNameResponse: - """ - Get an attribute by name with language-dependent properties. - - Args: - attribute_name: Name of the attribute - lang: Language code (optional) - - Returns: - AttributeGetByNameResponse with attribute details and languageDependents - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"attributes/name/{attribute_name}", - AttributeGetByNameResponse, - params=params if params else None, - ) - - 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( # type: ignore[attr-defined] - "GET", - f"attribute/groups/hierarchy/{attribute_group_id}", - AttributeGroupHierarchyResponse, - params=params, - ) diff --git a/elytra_client/rest_api/client/base.py b/elytra_client/rest_api/client/base.py deleted file mode 100644 index e999849..0000000 --- a/elytra_client/rest_api/client/base.py +++ /dev/null @@ -1,152 +0,0 @@ -"""Base client for the Lobster PIM Legacy REST API.""" - -from __future__ import annotations - -from typing import Any, Dict, Optional, Type, TypeVar -from urllib.parse import urljoin - -import requests -from pydantic import BaseModel, ValidationError - -from ...exceptions import ( - ElytraAPIError, - ElytraAuthenticationError, - ElytraNotFoundError, - ElytraValidationError, -) -from ..auth import AuthMethod, RestApiAuth - -T = TypeVar("T", bound=BaseModel) - - -class LobsterRestApiClientBase: - """ - Base client for the Lobster PIM Legacy REST API. - - Provides core infrastructure for authentication, HTTP requests, and response handling. - Subclasses should extend this with domain-specific methods via mixins. - - Args: - base_url: The base URL of the Lobster PIM server (e.g., http://localhost:8080) - auth: RestApiAuth instance for authentication - timeout: Request timeout in seconds (default: 30) - """ - - def __init__( - self, - base_url: str, - auth: RestApiAuth, - timeout: int = 30, - ): - """Initialize the Lobster REST API client""" - self.base_url = base_url.rstrip("/") - self.auth = auth - self.timeout = timeout - self.session = requests.Session() - self._setup_headers() - - def _setup_headers(self) -> None: - """Set up request headers including authentication""" - self.session.headers.update( - { - "Content-Type": "application/json", - "Accept": "application/json", - } - ) - self.session.headers.update(self.auth.get_auth_header()) - - def _handle_response( - self, - response: requests.Response, - expected_model: Type[T], - ) -> T: - """ - Handle API response and parse into Pydantic model. - - Args: - response: Response from requests - expected_model: Pydantic model to deserialize into - - Returns: - Parsed response as Pydantic model - - Raises: - ElytraAuthenticationError: If authentication fails - ElytraNotFoundError: If resource not found - ElytraAPIError: For other API errors - ElytraValidationError: If response validation fails - """ - if response.status_code == 401: - raise ElytraAuthenticationError("Authentication failed") - elif response.status_code == 404: - raise ElytraNotFoundError("Resource not found") - elif response.status_code == 429: - raise ElytraAPIError("Too many requests - rate limit exceeded") - elif response.status_code >= 400: - try: - error_data = response.json() - error_msg = error_data.get("error") or error_data.get("message", response.text) - except Exception: - error_msg = response.text - raise ElytraAPIError(f"API error {response.status_code}: {error_msg}") - - try: - data = response.json() - except Exception as e: - raise ElytraAPIError(f"Failed to parse response as JSON: {str(e)}") - - try: - return expected_model.model_validate(data) - except ValidationError as e: - raise ElytraValidationError(f"Response validation failed: {str(e)}") - - def _make_request( - self, - method: str, - endpoint: str, - expected_model: Type[T], - params: Optional[Dict[str, Any]] = None, - json_data: Optional[Dict[str, Any]] = None, - ) -> T: - """ - Make HTTP request to the REST API. - - Args: - method: HTTP method (GET, POST, etc.) - endpoint: API endpoint path - expected_model: Pydantic model for response - params: Query parameters - json_data: JSON request body - - Returns: - Parsed response as Pydantic model - """ - url = urljoin(self.base_url, f"/rest/{endpoint}") - - # Add authentication parameters for GET requests - if method.upper() == "GET" and self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: - if params is None: - params = {} - params.update(self.auth.get_url_parameters()) - - response = self.session.request( - method=method, - url=url, - params=params, - json=json_data, - timeout=self.timeout, - ) - - return self._handle_response(response, expected_model) - - def __enter__(self): - """Context manager entry""" - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - """Context manager exit""" - self.close() - - def close(self): - """Close the session""" - self.session.close() diff --git a/elytra_client/rest_api/client/jobs.py b/elytra_client/rest_api/client/jobs.py deleted file mode 100644 index 1d2ae3a..0000000 --- a/elytra_client/rest_api/client/jobs.py +++ /dev/null @@ -1,195 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Jobs mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, Optional - -from ..models import ( - JobControlRequest, - JobControlResponse, - JobDetailInfo, - JobExecutionResponse, - JobInfo, - JobOverviewResponse, -) - - -class JobsMixin: - """Mixin for job-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_job_overview(self) -> JobOverviewResponse: - """ - Get overview of all active jobs. - - Returns: - JobOverviewResponse containing list of active jobs - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - "job/overview", - JobOverviewResponse, - ) - - def get_job_html_overview(self) -> str: - """ - Get HTML overview of all active jobs. - - Returns: - HTML page content with job overview - """ - from urllib.parse import urljoin - - from ...exceptions import ElytraAPIError - from ..auth import AuthMethod - - url = urljoin(self.base_url, "/rest/job/overview") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.get(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to get job overview: {response.status_code}") - return response.text - - def get_all_active_jobs(self) -> JobOverviewResponse: - """ - Get list of all active jobs. - - Returns: - JobOverviewResponse containing list of all active jobs - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - "job/job_id", - JobOverviewResponse, - ) - - def get_job_detail(self, job_id: int) -> JobDetailInfo: - """ - Get details of a specific active job. - - Args: - job_id: ID of the job - - Returns: - JobDetailInfo with job details - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"job/job_id/{job_id}", - JobDetailInfo, - ) - - def execute_job( - self, - job_id: int, - ) -> JobExecutionResponse: - """ - Execute a job and get details of the started job. - - Args: - job_id: ID of the job to execute - - Returns: - JobExecutionResponse with execution details - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"job/execute/{job_id}", - JobExecutionResponse, - ) - - def get_running_job_instances(self) -> JobOverviewResponse: - """ - Get list of running job instances. - - Returns: - JobOverviewResponse containing list of running job instances - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - "job/runtime_id", - JobOverviewResponse, - ) - - def get_running_job_instance(self, runtime_id: str) -> JobOverviewResponse: - """ - Get details of a specific running job instance. - - Args: - runtime_id: Runtime ID of the job instance - - Returns: - JobOverviewResponse with instance details - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"job/runtime_id/{runtime_id}", - JobOverviewResponse, - ) - - def control_job( - self, - job_id: int, - action: str = "start", - additional_reference: Optional[str] = None, - parameters: Optional[Dict[str, Any]] = None, - queue_id: Optional[str] = None, - max_job_duration_seconds: Optional[int] = None, - ) -> JobControlResponse: - """ - Control a job using the control endpoint (POST). - - Supports starting jobs with parameter overrides and queue management. - - Args: - job_id: ID of the job to control - action: Action to perform (default: "start") - additional_reference: Optional reference for external tracking - parameters: Optional parameters to override job settings - queue_id: Optional queue ID for serialized execution - max_job_duration_seconds: Max duration in seconds (default 12 hours) - - Returns: - JobControlResponse with execution details - """ - request_body = { - "action": action, - "objectId": job_id, - "objectType": "job", - } - - # Add authentication credentials for POST - request_body.update(self.auth.get_json_body_params()) # type: ignore[attr-defined] - - if additional_reference: - request_body["additionalReference"] = additional_reference - - if parameters: - request_body["parameter"] = parameters - - if queue_id: - request_body["queueId"] = queue_id - - if max_job_duration_seconds: - request_body["maxJobDurationSeconds"] = max_job_duration_seconds - - return self._make_request( # type: ignore[attr-defined] - "POST", - "job/control", - JobControlResponse, - json_data=request_body, - ) diff --git a/elytra_client/rest_api/client/media.py b/elytra_client/rest_api/client/media.py deleted file mode 100644 index d027964..0000000 --- a/elytra_client/rest_api/client/media.py +++ /dev/null @@ -1,240 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Media mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional -from urllib.parse import urljoin - -from ...exceptions import ElytraAPIError -from ..auth import AuthMethod -from ..models import ( - MediaBulkCreateResponse, - MediaBulkUpdateResponse, - MediaFileResponse, - MediaListResponse, - SingleMediaResponse, -) - - -class MediaMixin: - """Mixin for media-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - 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( # type: ignore[attr-defined] - "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( # type: ignore[attr-defined] - "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( # type: ignore[attr-defined] - "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( # type: ignore[attr-defined] - "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( # type: ignore[attr-defined] - "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( # type: ignore[attr-defined] - "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}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - 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") # type: ignore[attr-defined] - - 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()) # type: ignore[attr-defined] - - response = self.session.post( # type: ignore[attr-defined] - url, - files=files, - data=data, - timeout=self.timeout, # type: ignore[attr-defined] - ) - - return self._handle_response(response, MediaFileResponse) # type: ignore[attr-defined] - - 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}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.get(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - 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}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete media file: {response.status_code}") diff --git a/elytra_client/rest_api/client/product_groups.py b/elytra_client/rest_api/client/product_groups.py deleted file mode 100644 index 81606ea..0000000 --- a/elytra_client/rest_api/client/product_groups.py +++ /dev/null @@ -1,193 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Product Groups mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional -from urllib.parse import urljoin - -from ...exceptions import ElytraAPIError -from ..auth import AuthMethod -from ..models import ( - ProductGroupBulkCreateResponse, - ProductGroupBulkUpdateResponse, - ProductGroupHierarchyResponse, - ProductGroupListResponse, - SingleProductGroupResponse, -) - - -class ProductGroupsMixin: - """Mixin for product group-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_all_product_groups( - self, - lang: Optional[str] = None, - page: int = 1, - limit: int = 10, - ) -> ProductGroupListResponse: - """ - Get all product groups. - - Args: - lang: Language code (optional) - page: Page number (default: 1) - limit: Number of groups per page (default: 10) - - Returns: - ProductGroupListResponse with paginated list of groups - """ - params: Dict[str, Any] = {"page": page, "limit": limit} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - "groups", - ProductGroupListResponse, - params=params, - ) - - def create_product_group(self, group_data: Dict[str, Any]) -> SingleProductGroupResponse: - """ - Create a new product group. - - Args: - group_data: Product group data - - Returns: - SingleProductGroupResponse with created group - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "groups", - SingleProductGroupResponse, - json_data=group_data, - ) - - def update_product_group(self, group_data: Dict[str, Any]) -> ProductGroupListResponse: - """ - Update a product group. - - Args: - group_data: Updated group data (must include 'id') - - Returns: - ProductGroupListResponse with updated group info - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "groups", - ProductGroupListResponse, - json_data=group_data, - ) - - def get_product_group_by_id( - self, group_id: int, lang: Optional[str] = None - ) -> SingleProductGroupResponse: - """ - Get a product group by ID. - - Args: - group_id: ID of the product group - lang: Language code (optional) - - Returns: - SingleProductGroupResponse with group details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"groups/{group_id}", - SingleProductGroupResponse, - params=params if params else None, - ) - - def delete_product_group(self, group_id: int) -> None: - """ - Delete a product group by ID. - - Args: - group_id: ID of the product group to delete - """ - url = urljoin(self.base_url, f"/rest/groups/{group_id}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete product group: {response.status_code}") - - def create_multiple_product_groups( - self, groups_list: List[Dict[str, Any]] - ) -> ProductGroupBulkCreateResponse: - """ - Create multiple product groups in bulk. - - Args: - groups_list: List of product group data - - Returns: - ProductGroupBulkCreateResponse with created groups - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "groups/bulk", - ProductGroupBulkCreateResponse, - json_data=groups_list, - ) - - def update_multiple_product_groups( - self, groups_list: List[Dict[str, Any]] - ) -> ProductGroupBulkUpdateResponse: - """ - Update multiple product groups in bulk. - - Args: - groups_list: List of product group data to update (each must include 'id') - - Returns: - ProductGroupBulkUpdateResponse with updated groups - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "groups/bulk", - ProductGroupBulkUpdateResponse, - json_data=groups_list, - ) - - 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( # type: ignore[attr-defined] - "GET", - f"groups/{group_id}/hierarchy", - ProductGroupHierarchyResponse, - params=params, - ) diff --git a/elytra_client/rest_api/client/products.py b/elytra_client/rest_api/client/products.py deleted file mode 100644 index a148b65..0000000 --- a/elytra_client/rest_api/client/products.py +++ /dev/null @@ -1,212 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Products mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional -from urllib.parse import urljoin - -from ...exceptions import ElytraAPIError -from ..auth import AuthMethod -from ..models import ( - ProductBulkCreateResponse, - ProductBulkUpdateResponse, - ProductHierarchyResponse, - ProductListResponse, - SingleProductResponse, -) - - -class ProductsMixin: - """Mixin for product-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_all_products( - self, - lang: Optional[str] = None, - page: int = 1, - limit: int = 10, - group_id: Optional[int] = None, - ) -> ProductListResponse: - """ - Get all products with optional group filter. - - Args: - lang: Language code (optional) - page: Page number (default: 1) - limit: Number of products per page (default: 10) - group_id: Optional product group ID to filter products - - Returns: - ProductListResponse with paginated list of products - """ - params: Dict[str, Any] = {"page": page, "limit": limit} - if lang: - params["lang"] = lang - if group_id: - params["groupId"] = group_id - - return self._make_request( # type: ignore[attr-defined] - "GET", - "products", - ProductListResponse, - params=params, - ) - - def create_product(self, product_data: Dict[str, Any]) -> SingleProductResponse: - """ - Create a new product. - - Args: - product_data: Product data - - Returns: - SingleProductResponse with created product - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "products", - SingleProductResponse, - json_data=product_data, - ) - - def update_product(self, product_data: Dict[str, Any]) -> ProductListResponse: - """ - Update a product. - - Args: - product_data: Updated product data (must include 'id') - - Returns: - ProductListResponse with updated product info - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "products", - ProductListResponse, - json_data=product_data, - ) - - def create_multiple_products( - self, products_list: List[Dict[str, Any]] - ) -> ProductBulkCreateResponse: - """ - Create multiple products in bulk. - - Args: - products_list: List of product data - - Returns: - ProductBulkCreateResponse with created products - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "products/bulk", - ProductBulkCreateResponse, - json_data=products_list, - ) - - def update_multiple_products( - self, products_list: List[Dict[str, Any]] - ) -> ProductBulkUpdateResponse: - """ - Update multiple products in bulk. - - Args: - products_list: List of product data to update (each must include 'id') - - Returns: - ProductBulkUpdateResponse with updated products - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "products/bulk", - ProductBulkUpdateResponse, - json_data=products_list, - ) - - def get_product_by_id( - self, product_id: int, lang: Optional[str] = None - ) -> SingleProductResponse: - """ - Get a product by ID. - - Args: - product_id: ID of the product - lang: Language code (optional) - - Returns: - SingleProductResponse with product details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"products/{product_id}", - SingleProductResponse, - params=params if params else None, - ) - - def delete_product(self, product_id: int) -> None: - """ - Delete a product by ID. - - Args: - product_id: ID of the product to delete - """ - url = urljoin(self.base_url, f"/rest/products/{product_id}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete product: {response.status_code}") - - def product_operation(self, operation_data: Dict[str, Any]) -> ProductListResponse: - """ - Perform bulk operations on products (copy, move, link, copy-structure). - - Args: - operation_data: Operation details including operation type, source, target, etc. - - Returns: - ProductListResponse with affected products - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "products/operation", - ProductListResponse, - json_data=operation_data, - ) - - 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( # type: ignore[attr-defined] - "GET", - f"products/{product_id}/hierarchy", - ProductHierarchyResponse, - params=params, - ) diff --git a/elytra_client/rest_api/client/protocols.py b/elytra_client/rest_api/client/protocols.py deleted file mode 100644 index 99708cf..0000000 --- a/elytra_client/rest_api/client/protocols.py +++ /dev/null @@ -1,134 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Protocols mixin for the Lobster REST API client.""" - -from typing import Optional -from urllib.parse import urljoin - -from ..models import ProtocolCategoryListResponse, ProtocolInfo, ProtocolListResponse - - -class ProtocolsMixin: - """Mixin for protocol-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_protocols(self, limit: int = 50) -> ProtocolListResponse: - """ - Get the last N protocols. - - Args: - limit: Number of protocols to retrieve (default: 50) - - Returns: - ProtocolListResponse containing list of protocols - """ - response = self.session.get( # type: ignore[attr-defined] - urljoin(self.base_url, "/rest/protocol"), # type: ignore[attr-defined] - params={"limit": limit, **self.auth.get_url_parameters()}, # type: ignore[attr-defined] - timeout=self.timeout, # type: ignore[attr-defined] - ) - return self._handle_response(response, ProtocolListResponse) # type: ignore[attr-defined] - - def get_protocol(self, protocol_id: str) -> ProtocolInfo: - """ - Get details of a specific protocol. - - Args: - protocol_id: ID of the protocol - - Returns: - ProtocolInfo with protocol details - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"protocol/{protocol_id}", - ProtocolInfo, - ) - - def get_protocol_by_job_id(self, job_id: int) -> ProtocolListResponse: - """ - Get all protocols for a specific job ID. - - Args: - job_id: ID of the job - - Returns: - ProtocolListResponse containing protocols for the job - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"protocol/job/{job_id}", - ProtocolListResponse, - ) - - def get_protocol_by_runtime_id(self, runtime_id: str) -> ProtocolInfo: - """ - Get protocol for a specific job instance runtime ID. - - Args: - runtime_id: Runtime ID of the job instance - - Returns: - ProtocolInfo with protocol details - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"protocol/job/{runtime_id}", - ProtocolInfo, - ) - - def get_protocol_by_additional_reference(self, reference: str) -> ProtocolListResponse: - """ - Get all protocols for a specific additional reference. - - Args: - reference: Additional reference value - - Returns: - ProtocolListResponse containing protocols for the reference - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"protocol/job/{reference}", - ProtocolListResponse, - ) - - def get_all_protocol_categories(self) -> ProtocolCategoryListResponse: - """ - Get all available protocol categories. - - Returns: - ProtocolCategoryListResponse containing list of categories - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - "protocol/category", - ProtocolCategoryListResponse, - ) - - def get_protocol_by_category(self, category_id: str, limit: int = 50) -> ProtocolListResponse: - """ - Get the last N protocols from a specific category. - - Args: - category_id: ID of the protocol category - limit: Number of protocols to retrieve (default: 50) - - Returns: - ProtocolListResponse containing protocols for the category - """ - return self._make_request( # type: ignore[attr-defined] - "GET", - f"protocol/category/{category_id}", - ProtocolListResponse, - params={"limit": limit}, - ) diff --git a/elytra_client/rest_api/client/text.py b/elytra_client/rest_api/client/text.py deleted file mode 100644 index f7e0780..0000000 --- a/elytra_client/rest_api/client/text.py +++ /dev/null @@ -1,164 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Text mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional - -from ..models import ( - SingleTextResponse, - TextBulkCreateResponse, - TextBulkUpdateResponse, - TextListResponse, -) - - -class TextMixin: - """Mixin for text-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_all_texts( - self, lang: Optional[str] = None, page: int = 1, limit: int = 10 - ) -> TextListResponse: - """ - Get all text entries with pagination. - - Args: - lang: Language code (optional) - page: Page number (default: 1) - limit: Number of text items per page (default: 10) - - Returns: - TextListResponse containing paginated list of text items - """ - params: Dict[str, Any] = {"page": page, "limit": limit} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - "text", - TextListResponse, - params=params, - ) - - def get_text_by_id(self, text_id: int, lang: Optional[str] = None) -> SingleTextResponse: - """ - Get a text entry by ID. - - Args: - text_id: ID of the text entry - lang: Language code (optional) - - Returns: - SingleTextResponse with text details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"text/{text_id}", - SingleTextResponse, - params=params if params else None, - ) - - def create_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: - """ - Create a new text entry. - - Args: - text_data: Text entry data - - Returns: - SingleTextResponse with created text - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "text", - SingleTextResponse, - json_data=text_data, - ) - - def update_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: - """ - Update a text entry. - - Args: - text_data: Updated text entry data (must include 'id') - - Returns: - SingleTextResponse with updated text - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "text", - SingleTextResponse, - json_data=text_data, - ) - - def create_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkCreateResponse: - """ - Create multiple text entries in bulk. - - Args: - texts_list: List of text entry data - - Returns: - TextBulkCreateResponse with created text entries - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "text/bulk", - TextBulkCreateResponse, - json_data=texts_list, - ) - - def update_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkUpdateResponse: - """ - Update multiple text entries in bulk. - - Args: - texts_list: List of text entry data to update (each must include 'id') - - Returns: - TextBulkUpdateResponse with updated text entries - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "text/bulk", - TextBulkUpdateResponse, - json_data=texts_list, - ) - - def delete_text(self, text_id: int) -> None: - """ - Delete a text entry by ID. - - Args: - text_id: ID of the text entry to delete - """ - from urllib.parse import urljoin - - from ...exceptions import ElytraAPIError - from ..auth import AuthMethod - - url = urljoin(self.base_url, f"/rest/text/{text_id}") # type: ignore[attr-defined] - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete text: {response.status_code}") diff --git a/elytra_client/rest_api/client/tree_groups.py b/elytra_client/rest_api/client/tree_groups.py deleted file mode 100644 index f65d4d9..0000000 --- a/elytra_client/rest_api/client/tree_groups.py +++ /dev/null @@ -1,209 +0,0 @@ -# mypy: disable-error-code="attr-defined, no-any-return" -"""Tree Groups mixin for the Lobster REST API client.""" - -from __future__ import annotations - -from typing import Any, Dict, List, Optional -from urllib.parse import urljoin - -from ...exceptions import ElytraAPIError, ElytraValidationError -from ..auth import AuthMethod -from ..models import ( - SingleTreeGroupResponse, - TreeGroupBulkCreateResponse, - TreeGroupBulkUpdateResponse, - TreeGroupHierarchyResponse, - TreeGroupListResponse, -) - - -class TreeGroupsMixin: - """Mixin for tree group-related operations. - # type: ignore[attr-defined] # Mixin classes inherit _make_request/_handle_response from base - - Expects to be mixed with LobsterRestApiClientBase or a compatible base class - that provides: base_url, auth, timeout, session, _make_request(), _handle_response(). - """ - - # Type hints for mixed-in attributes from base class (docstring only for clarity) - # base_url: str - from LobsterRestApiClientBase - # auth: Any - from LobsterRestApiClientBase - # timeout: int - from LobsterRestApiClientBase - # session: requests.Session - from LobsterRestApiClientBase - - def get_all_tree_groups( - self, - lang: Optional[str] = None, - page: int = 1, - limit: int = 10, - ) -> TreeGroupListResponse: - """ - Get all tree groups. - - Args: - lang: Language code (optional) - page: Page number (default: 1) - limit: Number of groups per page (default: 10) - - Returns: - TreeGroupListResponse with paginated list of tree groups - """ - params: Dict[str, Any] = {"page": page, "limit": limit} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - "tree/groups", - TreeGroupListResponse, - params=params, - ) - - def create_tree_group(self, group_data: Dict[str, Any]) -> SingleTreeGroupResponse: - """ - Create a new tree group. - - Args: - group_data: Tree group data - - Returns: - SingleTreeGroupResponse with created tree group - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "tree/groups", - SingleTreeGroupResponse, - json_data=group_data, - ) - - def update_tree_group(self, group_data: Dict[str, Any]) -> SingleTreeGroupResponse: - """ - Update a tree group. - - Args: - group_data: Updated tree group data (must include 'id') - - Returns: - SingleTreeGroupResponse with updated tree group - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "tree/groups", - SingleTreeGroupResponse, - json_data=group_data, - ) - - def get_tree_group_by_id( - self, tree_group_id: int, lang: Optional[str] = None - ) -> SingleTreeGroupResponse: - """ - Get a tree group by ID. - - Args: - tree_group_id: ID of the tree group - lang: Language code (optional) - - Returns: - SingleTreeGroupResponse with tree group details - """ - params: Dict[str, Any] = {} - if lang: - params["lang"] = lang - - return self._make_request( # type: ignore[attr-defined] - "GET", - f"tree/groups/{tree_group_id}", - SingleTreeGroupResponse, - params=params if params else None, - ) - - def delete_tree_group(self, tree_group_id: int) -> None: - """ - Delete a tree group by ID. - - Args: - tree_group_id: ID of the tree group to delete - """ - url = urljoin( - self.base_url, f"/rest/tree/groups/{tree_group_id}" # type: ignore[attr-defined] - ) - params = None - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params = self.auth.get_url_parameters() # type: ignore[attr-defined] - - response = self.session.delete(url, params=params, timeout=self.timeout) # type: ignore[attr-defined] - if response.status_code >= 400: - raise ElytraAPIError(f"Failed to delete tree group: {response.status_code}") - - def create_multiple_tree_groups( - self, groups_list: List[Dict[str, Any]] - ) -> TreeGroupBulkCreateResponse: - """ - Create multiple tree groups in bulk. - - Args: - groups_list: List of tree group data - - Returns: - TreeGroupBulkCreateResponse with created tree groups - """ - return self._make_request( # type: ignore[attr-defined] - "POST", - "tree/groups/bulk", - TreeGroupBulkCreateResponse, - json_data=groups_list, - ) - - def update_multiple_tree_groups( - self, groups_list: List[Dict[str, Any]] - ) -> TreeGroupBulkUpdateResponse: - """ - Update multiple tree groups in bulk. - - Args: - groups_list: List of tree group data to update (each must include 'id') - - Returns: - TreeGroupBulkUpdateResponse with updated tree groups - """ - return self._make_request( # type: ignore[attr-defined] - "PATCH", - "tree/groups/bulk", - TreeGroupBulkUpdateResponse, - json_data=groups_list, - ) - - 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") # type: ignore[attr-defined] - - if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: # type: ignore[attr-defined] - params.update(self.auth.get_url_parameters()) # type: ignore[attr-defined] - - response = self.session.request( # type: ignore[attr-defined] - method="GET", - url=url, - params=params, - timeout=self.timeout, # type: ignore[attr-defined] - ) - - 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)}") diff --git a/elytra_client/rest_api/models/jobs.py b/elytra_client/rest_api/models.py similarity index 58% rename from elytra_client/rest_api/models/jobs.py rename to elytra_client/rest_api/models.py index 9bf0923..0bbabe9 100644 --- a/elytra_client/rest_api/models/jobs.py +++ b/elytra_client/rest_api/models.py @@ -1,4 +1,4 @@ -"""Job management models""" +"""Models for the Lobster PIM Legacy REST API""" from typing import Any, Dict, List, Optional @@ -7,7 +7,6 @@ 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") @@ -23,14 +22,12 @@ 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") @@ -38,7 +35,6 @@ 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") @@ -48,15 +44,12 @@ 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") @@ -68,7 +61,9 @@ 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)" ) @@ -76,9 +71,59 @@ 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") messages: List[str] = Field(default_factory=list, description="List of messages") warnings: List[str] = Field(default_factory=list, description="List of warnings") + + +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") + + +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") + runtimeId: Optional[str] = Field(None, description="Runtime ID of the job execution") + jobIdentifier: Optional[str] = Field(None, description="Job identifier") + status: Optional[str] = Field(None, description="Status of the job") + startTime: Optional[str] = Field(None, description="Start time of execution") + 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" + ) + + +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") + + +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") + + +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") diff --git a/elytra_client/rest_api/models/__init__.py b/elytra_client/rest_api/models/__init__.py deleted file mode 100644 index 48c93eb..0000000 --- a/elytra_client/rest_api/models/__init__.py +++ /dev/null @@ -1,194 +0,0 @@ -"""Models package for Lobster PIM Legacy REST API""" - -# Shared models -# Attribute group models -from .attribute_groups import ( - AttributeGroupBulkCreateResponse, - AttributeGroupHierarchyNode, - AttributeGroupHierarchyResponse, - AttributeGroupListResponse, - AttributeGroupValidFor, - SingleAttributeGroupResponse, - SingleNewAttributeGroupRequestBody, - SingleUpdateAttributeGroupRequestBody, -) - -# Attribute models -from .attributes import ( - AttributeBulkCreateResponse, - AttributeBulkUpdateResponse, - AttributeGetByNameResponse, - AttributeListResponse, - SimpleAttributeResponse, - SingleNewAttributeRequestBody, - SingleUpdateAttributeRequestBody, -) - -# Hierarchy models -from .hierarchy import HierarchyNode - -# Job models -from .jobs import ( - JobControlRequest, - JobControlResponse, - JobDetailInfo, - JobExecutionResponse, - JobInfo, - JobOverviewResponse, -) - -# Media models -from .media import ( - MediaBulkCreateResponse, - MediaBulkUpdateResponse, - MediaFileResponse, - MediaListResponse, - SingleMediaResponse, - SingleNewMediaRequestBody, - SingleUpdateMediaRequestBody, -) - -# Product group models -from .product_groups import ( - ProductGroupBulkCreateResponse, - ProductGroupBulkUpdateResponse, - ProductGroupHierarchyNode, - ProductGroupHierarchyResponse, - ProductGroupListResponse, - SingleNewProductGroupRequestBody, - SingleProductGroupResponse, - SingleUpdateProductGroupRequestBody, -) - -# Product models -from .products import ( - ProductAttributeResponse, - ProductBulkCreateResponse, - ProductBulkUpdateResponse, - ProductHierarchyNode, - ProductHierarchyResponse, - ProductListResponse, - ProductOperationRequestBody, - SingleNewProductRequestBody, - SingleProductResponse, - SingleUpdateProductRequestBody, -) - -# Protocol models -from .protocols import ( - ProtocolCategoryInfo, - ProtocolCategoryListResponse, - ProtocolEntry, - ProtocolInfo, - ProtocolListResponse, -) -from .shared import AttributeResponse, ErrorResponse, PaginationLinks - -# Text models -from .text import ( - SingleNewTextRequestBody, - SingleTextResponse, - SingleUpdateTextRequestBody, - TextBulkCreateResponse, - TextBulkUpdateResponse, - TextContentRequestBody, - TextContentResponse, - TextListResponse, -) - -# Tree group models -from .tree_groups import ( - SingleNewTreeGroupRequestBody, - SingleTreeGroupResponse, - SingleUpdateTreeGroupRequestBody, - TreeGroupBulkCreateResponse, - TreeGroupBulkUpdateResponse, - TreeGroupHierarchyNode, - TreeGroupHierarchyResponse, - TreeGroupListResponse, -) - -__all__ = [ - # Shared - "AttributeResponse", - "ErrorResponse", - "PaginationLinks", - # Hierarchy - "HierarchyNode", - # Jobs - "JobControlRequest", - "JobControlResponse", - "JobDetailInfo", - "JobExecutionResponse", - "JobInfo", - "JobOverviewResponse", - # Protocols - "ProtocolCategoryInfo", - "ProtocolCategoryListResponse", - "ProtocolEntry", - "ProtocolInfo", - "ProtocolListResponse", - # Media - "MediaBulkCreateResponse", - "MediaBulkUpdateResponse", - "MediaFileResponse", - "MediaListResponse", - "SingleMediaResponse", - "SingleNewMediaRequestBody", - "SingleUpdateMediaRequestBody", - # Products - "ProductAttributeResponse", - "ProductBulkCreateResponse", - "ProductBulkUpdateResponse", - "ProductHierarchyNode", - "ProductHierarchyResponse", - "ProductListResponse", - "ProductOperationRequestBody", - "SingleNewProductRequestBody", - "SingleProductResponse", - "SingleUpdateProductRequestBody", - # Product groups - "ProductGroupBulkCreateResponse", - "ProductGroupBulkUpdateResponse", - "ProductGroupHierarchyNode", - "ProductGroupHierarchyResponse", - "ProductGroupListResponse", - "SingleNewProductGroupRequestBody", - "SingleProductGroupResponse", - "SingleUpdateProductGroupRequestBody", - # Tree groups - "SingleNewTreeGroupRequestBody", - "SingleTreeGroupResponse", - "SingleUpdateTreeGroupRequestBody", - "TreeGroupBulkCreateResponse", - "TreeGroupBulkUpdateResponse", - "TreeGroupHierarchyNode", - "TreeGroupHierarchyResponse", - "TreeGroupListResponse", - # Attributes - "AttributeBulkCreateResponse", - "AttributeBulkUpdateResponse", - "AttributeGetByNameResponse", - "AttributeListResponse", - "SimpleAttributeResponse", - "SingleNewAttributeRequestBody", - "SingleUpdateAttributeRequestBody", - # Attribute groups - "AttributeGroupBulkCreateResponse", - "AttributeGroupHierarchyNode", - "AttributeGroupHierarchyResponse", - "AttributeGroupListResponse", - "AttributeGroupValidFor", - "SingleAttributeGroupResponse", - "SingleNewAttributeGroupRequestBody", - "SingleUpdateAttributeGroupRequestBody", - # Text - "SingleNewTextRequestBody", - "SingleTextResponse", - "SingleUpdateTextRequestBody", - "TextBulkCreateResponse", - "TextBulkUpdateResponse", - "TextContentRequestBody", - "TextContentResponse", - "TextListResponse", -] diff --git a/elytra_client/rest_api/models/attribute_groups.py b/elytra_client/rest_api/models/attribute_groups.py deleted file mode 100644 index 6bbcdc5..0000000 --- a/elytra_client/rest_api/models/attribute_groups.py +++ /dev/null @@ -1,123 +0,0 @@ -"""Attribute group models""" - -from typing import Dict, List, Optional - -from pydantic import BaseModel, Field - - -class AttributeGroupValidFor(BaseModel): - """Valid object types for an attribute group""" - - product: Optional[bool] = Field(None, description="Valid for products") - productGroup: Optional[bool] = Field(None, description="Valid for product groups") - media: Optional[bool] = Field(None, description="Valid for media") - text: Optional[bool] = Field(None, description="Valid for texts") - - -class SingleAttributeGroupResponse(BaseModel): - """Complete attribute group descriptor""" - - id: int = Field(..., description="The ID of the attribute group") - name: str = Field(..., description="The independent name of the attribute group") - parentId: Optional[int] = Field(None, description="The ID of the parent attribute group") - validForObjectTypes: Optional[AttributeGroupValidFor] = Field( - None, description="Valid object types for this attribute group" - ) - isTemplate: bool = Field(False, description="Whether the attribute group is a template") - templateId: Optional[int] = Field(None, description="The ID of the template attribute group") - clientId: int = Field(..., description="The ID of the client") - createdAt: Optional[str] = Field( - None, description="The date and time the attribute group was created" - ) - createdByUserId: Optional[int] = Field( - None, description="The ID of user who created the attribute group" - ) - modifiedAt: Optional[str] = Field( - None, description="The date and time the attribute group was modified" - ) - modifiedByUserId: Optional[int] = Field( - None, description="The ID of user who modified the attribute group" - ) - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the attribute group name" - ) - - -class SingleNewAttributeGroupRequestBody(BaseModel): - """Request body for creating a new attribute group""" - - name: str = Field(..., description="The independent name of the attribute group") - parentId: Optional[int] = Field(None, description="The ID of the parent attribute group") - validForObjectTypes: Optional[AttributeGroupValidFor] = Field( - None, description="Valid object types for this attribute group" - ) - isTemplate: Optional[bool] = Field( - False, description="Whether the attribute group is a template" - ) - templateId: Optional[int] = Field(None, description="The ID of the template attribute group") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the attribute group name" - ) - - -class SingleUpdateAttributeGroupRequestBody(BaseModel): - """Request body for updating an attribute group""" - - id: Optional[int] = Field(None, description="The ID of the attribute group") - name: Optional[str] = Field(None, description="The independent name of the attribute group") - newName: Optional[str] = Field( - None, description="The new independent name of the attribute group" - ) - parentId: Optional[int] = Field(None, description="The ID of the parent attribute group") - validForObjectTypes: Optional[AttributeGroupValidFor] = Field( - None, description="Valid object types for this attribute group" - ) - isTemplate: Optional[bool] = Field( - None, description="Whether the attribute group is a template" - ) - templateId: Optional[int] = Field(None, description="The ID of the template attribute group") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the attribute group name" - ) - - -class AttributeGroupListResponse(BaseModel): - """Paginated response containing multiple attribute groups""" - - items: List[SingleAttributeGroupResponse] = Field(..., description="List of attribute groups") - total: int = Field(..., description="The total number of attribute groups") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of attribute groups per page") - links: Optional[Dict[str, Optional[str]]] = Field(None, description="Pagination links") - - -class AttributeGroupBulkCreateResponse(BaseModel): - """Response from bulk attribute group creation""" - - items: List[SingleAttributeGroupResponse] = Field( - ..., description="The created attribute groups" - ) - totalItemsCreated: int = Field(..., description="The total number of attribute groups created") - - -class AttributeGroupHierarchyNode(BaseModel): - """A node in the attribute group hierarchy""" - - 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 (attribute-group, attribute-group-template, attribute-group-derived-template, attribute)", - ) - children: List["AttributeGroupHierarchyNode"] = Field( - default_factory=list, description="The immediate children of the node" - ) - - -AttributeGroupHierarchyNode.model_rebuild() - - -class AttributeGroupHierarchyResponse(AttributeGroupHierarchyNode): - """Attribute group hierarchy response""" - - pass diff --git a/elytra_client/rest_api/models/attributes.py b/elytra_client/rest_api/models/attributes.py deleted file mode 100644 index addb6b7..0000000 --- a/elytra_client/rest_api/models/attributes.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Attribute models""" - -from typing import Any, Dict, List, Optional - -from pydantic import BaseModel, Field - - -class SimpleAttributeResponse(BaseModel): - """Simplified attribute definition""" - - id: int = Field(..., description="The ID of the attribute") - name: str = Field(..., description="The independent name of the attribute") - description: Optional[str] = Field(None, description="The description of the attribute") - attributeType: Optional[str] = Field( - None, description="The type of attribute (normal, meta, internal)" - ) - type: Optional[str] = Field(None, description="The type of the attribute") - autoSync: Optional[str] = Field(None, description="The auto sync mode of the attribute") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the attribute name" - ) - - -class AttributeGetByNameResponse(SimpleAttributeResponse): - """Attribute response when fetching by name""" - - languageDependents: Optional[Dict[str, Any]] = Field( - None, description="Language-dependent properties" - ) - - -class SingleNewAttributeRequestBody(BaseModel): - """Request body for creating a new attribute""" - - name: str = Field(..., description="The independent name of the attribute") - type: str = Field(..., description="The type of the attribute") - description: Optional[str] = Field(None, description="The description of the attribute") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the attribute name" - ) - - -class SingleUpdateAttributeRequestBody(BaseModel): - """Request body for updating an attribute""" - - id: int = Field(..., description="The ID of the attribute") - name: Optional[str] = Field(None, description="The independent name of the attribute") - description: Optional[str] = Field(None, description="The description of the attribute") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the attribute name" - ) - - -class AttributeListResponse(BaseModel): - """Paginated response containing multiple attributes""" - - items: List[SimpleAttributeResponse] = Field(..., description="List of attributes") - total: int = Field(..., description="The total number of attributes") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of attributes per page") - links: Optional[Dict[str, Optional[str]]] = Field(None, description="Pagination links") - - -class AttributeBulkCreateResponse(BaseModel): - """Response from bulk attribute creation""" - - items: List[SimpleAttributeResponse] = Field(..., description="The created attributes") - totalItemsCreated: int = Field(..., description="The total number of attributes created") - - -class AttributeBulkUpdateResponse(BaseModel): - """Response from bulk attribute update""" - - items: List[SimpleAttributeResponse] = Field(..., description="The updated attributes") - totalItemsUpdated: int = Field(..., description="The total number of attributes updated") diff --git a/elytra_client/rest_api/models/hierarchy.py b/elytra_client/rest_api/models/hierarchy.py deleted file mode 100644 index 017f386..0000000 --- a/elytra_client/rest_api/models/hierarchy.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Hierarchy models for tree structures""" - -from typing import List - -from pydantic import BaseModel, Field - - -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() diff --git a/elytra_client/rest_api/models/media.py b/elytra_client/rest_api/models/media.py deleted file mode 100644 index c19b937..0000000 --- a/elytra_client/rest_api/models/media.py +++ /dev/null @@ -1,133 +0,0 @@ -"""Media models""" - -from typing import Any, Dict, List, Optional - -from pydantic import BaseModel, Field - -from .shared import AttributeResponse, PaginationLinks - - -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") diff --git a/elytra_client/rest_api/models/product_groups.py b/elytra_client/rest_api/models/product_groups.py deleted file mode 100644 index 4631ac3..0000000 --- a/elytra_client/rest_api/models/product_groups.py +++ /dev/null @@ -1,105 +0,0 @@ -"""Product group models""" - -from typing import Any, Dict, List, Optional - -from pydantic import BaseModel, Field - -from .products import ProductAttributeResponse - - -class SingleProductGroupResponse(BaseModel): - """Complete product group descriptor""" - - id: int = Field(..., description="The ID of the product group") - name: str = Field(..., description="The independent name of the product group") - type: Optional[str] = Field(None, description="The type of product group") - parentId: Optional[int] = Field( - None, description="The ID of the parent product group or tree group" - ) - objectStatus: Optional[str] = Field(None, description="The status of the object") - attributeGroupId: Optional[int] = Field(None, description="The ID of the attribute group") - clientId: int = Field(..., description="The ID of the client") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the product group name" - ) - attributes: List[ProductAttributeResponse] = Field( - default_factory=list, description="The attributes of the product group" - ) - - -class SingleNewProductGroupRequestBody(BaseModel): - """Request body for creating a new product group""" - - name: str = Field(..., description="The independent name of the product group") - type: Optional[str] = Field(None, description="The type of product group") - parentId: Optional[int] = Field( - None, description="The ID of the parent product group or tree group" - ) - attributeGroupId: Optional[int] = Field(None, description="The ID of the attribute group") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the product group name" - ) - attributes: Optional[List[Dict[str, Any]]] = Field( - None, description="The attributes of the product group" - ) - - -class SingleUpdateProductGroupRequestBody(BaseModel): - """Request body for updating a product group""" - - id: int = Field(..., description="The ID of the product group") - name: Optional[str] = Field(None, description="The independent name of the product group") - parentId: Optional[int] = Field(None, description="The ID of the parent product group") - attributeGroupId: Optional[int] = Field(None, description="The ID of the attribute group") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the product group name" - ) - attributes: Optional[List[Dict[str, Any]]] = Field( - None, description="The attributes of the product group" - ) - - -class ProductGroupListResponse(BaseModel): - """Paginated response containing multiple product groups""" - - items: List[SingleProductGroupResponse] = Field(..., description="List of product groups") - total: int = Field(..., description="The total number of product groups") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of product groups per page") - links: Optional[Dict[str, Optional[str]]] = Field(None, description="Pagination links") - - -class ProductGroupBulkCreateResponse(BaseModel): - """Response from bulk product group creation""" - - items: List[SingleProductGroupResponse] = Field(..., description="The created product groups") - totalItemsCreated: int = Field(..., description="The total number of product groups created") - - -class ProductGroupBulkUpdateResponse(BaseModel): - """Response from bulk product group update""" - - items: List[SingleProductGroupResponse] = Field(..., description="The updated product groups") - totalItemsUpdated: int = Field(..., description="The total number of product groups updated") - - -class ProductGroupHierarchyNode(BaseModel): - """A node in the product group hierarchy""" - - 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, text, media)" - ) - children: List["ProductGroupHierarchyNode"] = Field( - default_factory=list, description="The immediate children of the node" - ) - - -ProductGroupHierarchyNode.model_rebuild() - - -class ProductGroupHierarchyResponse(ProductGroupHierarchyNode): - """Product group hierarchy response""" - - pass diff --git a/elytra_client/rest_api/models/products.py b/elytra_client/rest_api/models/products.py deleted file mode 100644 index 2b3bf22..0000000 --- a/elytra_client/rest_api/models/products.py +++ /dev/null @@ -1,120 +0,0 @@ -"""Product models""" - -from typing import Any, Dict, List, Optional - -from pydantic import BaseModel, Field - - -class ProductOperationRequestBody(BaseModel): - """Request body for product operations (copy, move, link, copy-structure)""" - - operation: str = Field(..., description="Operation: copy, move, link, or copy-structure") - productId: int = Field(..., description="The ID of the product to perform operation on") - parentId: int = Field(..., description="The ID of the destination parent") - - -class ProductAttributeResponse(BaseModel): - """An attribute value associated with a product""" - - id: int = Field(..., description="The ID of the attribute") - 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 type of attribute (normal, meta, internal)") - type: str = Field(..., description="The attribute type") - value: Optional[str] = Field(None, description="The value of the attribute") - autoSync: Optional[str] = Field(None, description="The auto sync mode of the attribute") - languageCode: Optional[str] = Field(None, description="The language code of the attribute") - modified: Optional[str] = Field( - None, description="The date and time the attribute was modified" - ) - modifierByUserId: Optional[int] = Field( - None, description="The ID of user who modified the attribute" - ) - inherited: bool = Field(False, description="Whether the attribute is inherited") - - -class SingleProductResponse(BaseModel): - """Complete product descriptor""" - - id: int = Field(..., description="The ID of the product") - clientId: int = Field(..., description="The ID of the client") - productName: str = Field(..., description="The name of the product") - treeId: int = Field(..., description="The ID of the tree") - created: Optional[str] = Field(None, description="The date and time the product was created") - modified: Optional[str] = Field(None, description="The date and time the product was modified") - creatorUserId: Optional[int] = Field(None, description="The ID of user who created the product") - modifierUserId: Optional[int] = Field( - None, description="The ID of user who modified the product" - ) - objectStatus: Optional[str] = Field(None, description="The status of the object") - originalId: Optional[int] = Field(None, description="The ID of the original product") - attributes: List[ProductAttributeResponse] = Field( - default_factory=list, description="The attributes of the product" - ) - - -class SingleNewProductRequestBody(BaseModel): - """Request body for creating a new product""" - - productName: str = Field(..., description="The name of the product") - parentId: int = Field(..., description="The ID of the parent group or product") - attributeGroupId: int = Field(..., description="The ID of the attribute group") - attributes: Optional[List[Dict[str, Any]]] = Field( - None, description="The attributes of the product" - ) - - -class SingleUpdateProductRequestBody(BaseModel): - """Request body for updating a product""" - - id: int = Field(..., description="The ID of the product") - productName: Optional[str] = Field(None, description="The name of the product") - parentId: Optional[int] = Field(None, description="The ID of the parent group or product") - attributeGroupId: Optional[int] = Field(None, description="The ID of the attribute group") - attributes: Optional[List[Dict[str, Any]]] = Field( - None, description="The attributes of the product" - ) - - -class ProductListResponse(BaseModel): - """Paginated response containing multiple products""" - - items: List[SingleProductResponse] = Field(..., description="List of products") - total: int = Field(..., description="The total number of products") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of products per page") - links: Optional[Dict[str, Optional[str]]] = Field(None, description="Pagination links") - - -class ProductBulkCreateResponse(BaseModel): - """Response from bulk product creation""" - - items: List[SingleProductResponse] = Field(..., description="The created products") - totalItemsCreated: int = Field(..., description="The total number of products created") - - -class ProductBulkUpdateResponse(BaseModel): - """Response from bulk product update""" - - items: List[SingleProductResponse] = Field(..., description="The updated products") - totalItemsUpdated: int = Field(..., description="The total number of products updated") - - -class ProductHierarchyNode(BaseModel): - """A node in the product hierarchy""" - - 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, text, media)") - children: List["ProductHierarchyNode"] = Field( - default_factory=list, description="The immediate children of the node" - ) - - -ProductHierarchyNode.model_rebuild() - - -class ProductHierarchyResponse(ProductHierarchyNode): - """Product hierarchy response""" - - pass diff --git a/elytra_client/rest_api/models/protocols.py b/elytra_client/rest_api/models/protocols.py deleted file mode 100644 index b46d31c..0000000 --- a/elytra_client/rest_api/models/protocols.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Protocol/Log models""" - -from typing import List, Optional - -from pydantic import BaseModel, Field - - -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") - - -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") - runtimeId: Optional[str] = Field(None, description="Runtime ID of the job execution") - jobIdentifier: Optional[str] = Field(None, description="Job identifier") - status: Optional[str] = Field(None, description="Status of the job") - startTime: Optional[str] = Field(None, description="Start time of execution") - 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") - - -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") - - -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") - - -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") diff --git a/elytra_client/rest_api/models/shared.py b/elytra_client/rest_api/models/shared.py deleted file mode 100644 index e4911db..0000000 --- a/elytra_client/rest_api/models/shared.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Shared models and common types for the Lobster PIM Legacy REST API""" - -from typing import Any, Dict, List, Optional - -from pydantic import BaseModel, Field - - -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") - - -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") diff --git a/elytra_client/rest_api/models/text.py b/elytra_client/rest_api/models/text.py deleted file mode 100644 index e7f7173..0000000 --- a/elytra_client/rest_api/models/text.py +++ /dev/null @@ -1,125 +0,0 @@ -"""Text models""" - -from typing import Any, Dict, List, Optional - -from pydantic import BaseModel, Field - -from .shared import AttributeResponse - - -class TextContentResponse(BaseModel): - """Text content with metadata""" - - id: int = Field(..., description="The ID of the text content") - clientId: int = Field(..., description="The ID of the client") - languageCode: str = Field(..., description="The language code for this text content") - mimeType: str = Field(..., description="The MIME type of the text content") - content: str = Field(..., description="The text content") - contentLength: int = Field(..., description="The size of the text content in bytes") - workflowStatus: Optional[str] = Field( - None, description="The workflow status of the text content" - ) - workflowComment: Optional[str] = Field(None, description="Comments related to the workflow") - changedAt: Optional[str] = Field( - None, description="The date and time the text content was last changed" - ) - changedBy: Optional[int] = Field( - None, description="The ID of the user who last changed the text content" - ) - - -class TextContentRequestBody(BaseModel): - """Request body for text content""" - - mimeType: str = Field(..., description="The MIME type of the text content") - content: str = Field(..., description="The text content") - languageCode: str = Field(..., description="The language code for this text content") - workflowStatus: Optional[str] = Field( - None, description="The workflow status of the text content" - ) - workflowComment: Optional[str] = Field(None, description="Comments related to the workflow") - - -class SingleTextResponse(BaseModel): - """Complete text descriptor""" - - id: int = Field(..., description="The ID of the text descriptor") - name: str = Field(..., description="The name of the text") - treeId: Optional[int] = Field(None, description="The ID of the tree this text belongs to") - clientId: int = Field(..., description="The ID of the client") - originalId: Optional[int] = Field( - None, description="The ID of the original text if this is a copy" - ) - objectStatus: Optional[str] = Field(None, description="The status of the object") - userObjectStatus: Optional[int] = Field(None, description="User-defined object status") - createdAt: Optional[str] = Field(None, description="The date and time the text was created") - createdBy: Optional[int] = Field(None, description="The ID of the user who created the text") - modifiedAt: Optional[str] = Field( - None, description="The date and time the text was last modified" - ) - modifiedBy: Optional[int] = Field( - None, description="The ID of the user who last modified the text" - ) - contents: List[TextContentResponse] = Field( - default_factory=list, description="The text contents for different languages" - ) - attributes: List[AttributeResponse] = Field( - default_factory=list, description="The attribute values of the text" - ) - - -class SingleNewTextRequestBody(BaseModel): - """Request body for creating a new text""" - - name: str = Field(..., description="The name of the text descriptor") - parentId: int = Field(..., description="The ID of the parent product or group") - languageCode: str = Field(..., description="The language code for the response") - textTypeId: int = Field(..., description="The ID of the text type") - contents: Optional[List[TextContentRequestBody]] = Field( - None, description="List of text contents for different languages" - ) - treeId: Optional[int] = Field(None, description="The ID of the tree this text belongs to") - attributeGroupId: Optional[int] = Field( - None, description="The ID of the attribute group to assign to this text" - ) - attributes: Optional[List[Dict[str, Any]]] = Field( - None, description="Optional attributes to set" - ) - - -class SingleUpdateTextRequestBody(BaseModel): - """Request body for updating a text""" - - id: int = Field(..., description="The ID of the text descriptor to update") - name: Optional[str] = Field(None, description="The name of the text descriptor") - userObjectStatus: Optional[int] = Field(None, description="User-defined object status") - textTypeId: Optional[int] = Field(None, description="The ID of the text type") - treeId: Optional[int] = Field(None, description="The ID of the tree this text belongs to") - contents: Optional[List[TextContentRequestBody]] = Field( - None, description="List of text contents to update" - ) - attributes: Optional[List[Dict[str, Any]]] = Field(None, description="Attributes to update") - - -class TextListResponse(BaseModel): - """Paginated response containing multiple texts""" - - items: List[SingleTextResponse] = Field(..., description="List of text items") - total: int = Field(..., description="The total number of text items") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of text items per page") - links: Optional[Dict[str, Optional[str]]] = Field(None, description="Pagination links") - - -class TextBulkCreateResponse(BaseModel): - """Response from bulk text creation""" - - items: List[SingleTextResponse] = Field(..., description="The created text items") - totalItemsCreated: int = Field(..., description="The total number of text items created") - - -class TextBulkUpdateResponse(BaseModel): - """Response from bulk text update""" - - items: List[SingleTextResponse] = Field(..., description="The updated text items") - totalItemsUpdated: int = Field(..., description="The total number of text items updated") diff --git a/elytra_client/rest_api/models/tree_groups.py b/elytra_client/rest_api/models/tree_groups.py deleted file mode 100644 index d4040f1..0000000 --- a/elytra_client/rest_api/models/tree_groups.py +++ /dev/null @@ -1,87 +0,0 @@ -"""Tree group models""" - -from typing import Dict, List, Optional - -from pydantic import BaseModel, Field - - -class SingleTreeGroupResponse(BaseModel): - """Complete tree group descriptor""" - - id: int = Field(..., description="The ID of the tree group") - name: str = Field(..., description="The independent name of the tree group") - parentId: Optional[int] = Field(None, description="The ID of the parent tree group") - clientId: int = Field(..., description="The ID of the client") - status: Optional[str] = Field(None, description="The status of the group (normal, internal)") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the tree group name" - ) - - -class SingleNewTreeGroupRequestBody(BaseModel): - """Request body for creating a new tree group""" - - name: str = Field(..., description="The name of the tree group") - parentId: int = Field(..., description="The ID of the parent tree group") - status: Optional[str] = Field( - "normal", description="The status of the tree group (normal, internal)" - ) - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the tree group name" - ) - - -class SingleUpdateTreeGroupRequestBody(BaseModel): - """Request body for updating a tree group""" - - id: int = Field(..., description="The ID of the tree group") - name: Optional[str] = Field(None, description="The name of the tree group") - parentId: Optional[int] = Field(None, description="The ID of the parent tree group") - status: Optional[str] = Field(None, description="The status of the tree group") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the tree group name" - ) - - -class TreeGroupListResponse(BaseModel): - """Paginated response containing multiple tree groups""" - - items: List[SingleTreeGroupResponse] = Field(..., description="List of tree groups") - total: int = Field(..., description="The total number of tree groups") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of tree groups per page") - links: Optional[Dict[str, Optional[str]]] = Field(None, description="Pagination links") - - -class TreeGroupBulkCreateResponse(BaseModel): - """Response from bulk tree group creation""" - - items: List[SingleTreeGroupResponse] = Field(..., description="The created tree groups") - totalItemsCreated: int = Field(..., description="The total number of tree groups created") - - -class TreeGroupBulkUpdateResponse(BaseModel): - """Response from bulk tree group update""" - - items: List[SingleTreeGroupResponse] = Field(..., description="The updated tree groups") - totalItemsUpdated: int = Field(..., description="The total number of tree groups updated") - - -class TreeGroupHierarchyNode(BaseModel): - """A node in the tree group hierarchy""" - - 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["TreeGroupHierarchyNode"] = Field( - default_factory=list, description="The immediate children of the node" - ) - - -TreeGroupHierarchyNode.model_rebuild() - - -class TreeGroupHierarchyResponse(TreeGroupHierarchyNode): - """Tree group hierarchy response""" - - pass diff --git a/pyproject.toml b/pyproject.toml index 7c3bc5c..22759dc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "elytra-pim-client" -version = "0.7.0" +version = "0.3.0" description = "A Pythonic client for the Elytra PIM API" readme = "README.md" requires-python = ">=3.9" @@ -44,8 +44,8 @@ Repository = "https://git.him-tools.de/HIM-public/elytra_client.git" Documentation = "https://www.elytra.ch/" Issues = "https://git.him-tools.de/HIM-public/elytra_client/issues" -[tool.setuptools.packages.find] -include = ["elytra_client*"] +[tool.setuptools] +packages = ["elytra_client"] [tool.black] line-length = 100 @@ -56,8 +56,7 @@ profile = "black" line_length = 100 [tool.mypy] -python_version = "3.10" +python_version = "0.3.0" warn_return_any = true warn_unused_configs = true disallow_untyped_defs = false -ignore_missing_imports = true diff --git a/tests/test_client.py b/tests/test_client.py index bc55293..20e64ce 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,11 +1,14 @@ """Tests for the Elytra PIM Client with Pydantic validation""" -from unittest.mock import Mock, patch - import pytest +from unittest.mock import Mock, patch from pydantic import ValidationError -from elytra_client import ElytraClient, SingleNewProductRequestBody, SingleProductResponse +from elytra_client import ( + ElytraClient, + SingleProductResponse, + SingleNewProductRequestBody, +) from elytra_client.exceptions import ElytraAuthenticationError, ElytraNotFoundError @@ -115,11 +118,11 @@ def test_create_product_with_pydantic(mock_request, client): mock_request.return_value = mock_response # Create with Pydantic model - new_product = SingleNewProductRequestBody( # type: ignore[arg-type] + new_product = SingleNewProductRequestBody( productName="NEW-PRODUCT-001", parentId=1, attributeGroupId=10, - ) + ) # type: ignore - validation happens automatically, so type checker should recognize this as valid result = client.create_product(new_product) @@ -131,20 +134,20 @@ def test_create_product_with_pydantic(mock_request, client): def test_pydantic_validation_on_creation(): """Test Pydantic validation on model creation""" # Valid model - valid_product = SingleNewProductRequestBody( # type: ignore[arg-type] + valid_product = SingleNewProductRequestBody( productName="VALID-PRODUCT", parentId=1, attributeGroupId=10, - ) + ) # type: ignore - validation happens automatically, so type checker should recognize this as valid assert valid_product.productName == "VALID-PRODUCT" # Invalid model - missing required field with pytest.raises(ValidationError): - SingleNewProductRequestBody( # type: ignore[arg-type] + SingleNewProductRequestBody( productName="INVALID-PRODUCT", # Missing parentId - required attributeGroupId=10, - ) + ) # type: ignore - this will raise a ValidationError, so type checker should recognize this as invalid @patch("elytra_client.client.requests.Session.request") @@ -153,9 +156,9 @@ def test_authentication_error(mock_request, client): mock_response = Mock() mock_response.status_code = 401 mock_response.text = "Unauthorized" - mock_request.return_value.raise_for_status.side_effect = __import__( - "requests" - ).exceptions.HTTPError(response=mock_response) + mock_request.return_value.raise_for_status.side_effect = ( + __import__("requests").exceptions.HTTPError(response=mock_response) + ) with pytest.raises(ElytraAuthenticationError): client.get_products() @@ -167,9 +170,9 @@ def test_not_found_error(mock_request, client): mock_response = Mock() mock_response.status_code = 404 mock_response.text = "Not Found" - mock_request.return_value.raise_for_status.side_effect = __import__( - "requests" - ).exceptions.HTTPError(response=mock_response) + mock_request.return_value.raise_for_status.side_effect = ( + __import__("requests").exceptions.HTTPError(response=mock_response) + ) with pytest.raises(ElytraNotFoundError): client.get_product(product_id=999) diff --git a/update_version.py b/update_version.py index 6cf194b..9e05bc4 100644 --- a/update_version.py +++ b/update_version.py @@ -27,28 +27,24 @@ def validate_version(version: str) -> bool: Returns: True if valid, False otherwise """ - pattern = r"^\d+\.\d+\.\d+(?:-[a-zA-Z0-9]+)?$" + pattern = r'^\d+\.\d+\.\d+(?:-[a-zA-Z0-9]+)?$' return bool(re.match(pattern, version)) -def update_file( - file_path: Path, old_pattern: str, new_version: str, multiline: bool = False -) -> bool: +def update_file(file_path: Path, old_pattern: str, new_version: str) -> bool: """Update version in a file. Args: file_path: Path to file to update old_pattern: Regex pattern to find version new_version: New version string - multiline: Whether to use multiline mode for regex Returns: True if successful, False otherwise """ try: content = file_path.read_text() - flags = re.MULTILINE if multiline else 0 - updated_content = re.sub(old_pattern, new_version, content, flags=flags) + updated_content = re.sub(old_pattern, new_version, content) if content == updated_content: print(f"✓ {file_path.name} already up-to-date") @@ -93,10 +89,11 @@ def main() -> int: # Update pyproject.toml if pyproject_path.exists(): - # Use word boundary to match 'version' but not 'python_version' - pattern = r'^version = "[^"]+"' + pattern = r'version = "[^"]+"' success &= update_file( - pyproject_path, pattern, f'version = "{new_version}"', multiline=True + pyproject_path, + pattern, + f'version = "{new_version}"' ) else: print(f"✗ {pyproject_path} not found") @@ -105,7 +102,11 @@ def main() -> int: # Update __init__.py if init_path.exists(): pattern = r'__version__ = "[^"]+"' - success &= update_file(init_path, pattern, f'__version__ = "{new_version}"') + success &= update_file( + init_path, + pattern, + f'__version__ = "{new_version}"' + ) else: print(f"✗ {init_path} not found") success = False