diff --git a/add_methods.py b/add_methods.py new file mode 100644 index 0000000..2847fdd --- /dev/null +++ b/add_methods.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +"""Script to add all missing methods to the REST client""" + +import re + +# Read the generated methods file +with open("generated_methods.py", "r") as f: + generated_content = f.read() + +# Extract the Python code from markdown backticks +match = re.search(r"```python\n(.*?)\n```", generated_content, re.DOTALL) +if not match: + print("ERROR: Could not find Python code block in generated_methods.py") + exit(1) + +methods_code = match.group(1) + +# Read the current client file +with open("elytra_client/rest_api/client.py", "r") as f: + client_content = f.read() + +# Find the position of "def close(self)" method +close_pos = client_content.find(" def close(self) -> None:") +if close_pos == -1: + print("ERROR: Could not find 'def close' method in client.py") + exit(1) + +# Insert the new methods before the close() method +new_client_content = client_content[:close_pos] + methods_code + "\n\n" + client_content[close_pos:] + +# Write back to the client file +with open("elytra_client/rest_api/client.py", "w") as f: + f.write(new_client_content) + +print(f"Successfully added {methods_code.count('def ')} new methods to the client") +print("Client file updated successfully") diff --git a/elytra_client/rest_api/client.py b/elytra_client/rest_api/client.py index 718af94..1103195 100644 --- a/elytra_client/rest_api/client.py +++ b/elytra_client/rest_api/client.py @@ -14,7 +14,12 @@ from ..exceptions import ( ) from .auth import AuthMethod, RestApiAuth from .models import ( + AttributeBulkCreateResponse, + AttributeBulkUpdateResponse, + AttributeGetByNameResponse, AttributeGroupHierarchyResponse, + AttributeGroupListResponse, + AttributeListResponse, AttributeResponse, JobControlRequest, JobControlResponse, @@ -26,16 +31,46 @@ from .models import ( MediaBulkUpdateResponse, MediaFileResponse, MediaListResponse, + ProductBulkCreateResponse, + ProductBulkUpdateResponse, + ProductGroupBulkCreateResponse, + ProductGroupBulkUpdateResponse, ProductGroupHierarchyResponse, + ProductGroupListResponse, ProductHierarchyResponse, + ProductListResponse, ProtocolCategoryInfo, ProtocolCategoryListResponse, ProtocolInfo, ProtocolListResponse, + SimpleAttributeResponse, + SingleAttributeGroupResponse, SingleMediaResponse, + SingleNewAttributeGroupRequestBody, + SingleNewAttributeRequestBody, SingleNewMediaRequestBody, + SingleNewProductGroupRequestBody, + SingleNewProductRequestBody, + SingleNewTextRequestBody, + SingleNewTreeGroupRequestBody, + SingleProductGroupResponse, + SingleProductResponse, + SingleTextResponse, + SingleTreeGroupResponse, + SingleUpdateAttributeGroupRequestBody, + SingleUpdateAttributeRequestBody, SingleUpdateMediaRequestBody, + SingleUpdateProductGroupRequestBody, + SingleUpdateProductRequestBody, + SingleUpdateTextRequestBody, + SingleUpdateTreeGroupRequestBody, + TextBulkCreateResponse, + TextBulkUpdateResponse, + TextListResponse, + TreeGroupBulkCreateResponse, + TreeGroupBulkUpdateResponse, TreeGroupHierarchyResponse, + TreeGroupListResponse, ) T = TypeVar("T", bound=BaseModel) @@ -745,6 +780,807 @@ class LobsterRestApiClient: params=params, ) + # ============= Product Endpoints ============= + + 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( + "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( + "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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete 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( + "POST", + "products/operation", + ProductListResponse, + json_data=operation_data, + ) + + # ============= Product Group Endpoints ============= + + 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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete product group: {response.status_code}") + + # ============= Tree Group Endpoints ============= + + 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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete tree group: {response.status_code}") + + # ============= Attribute Endpoints ============= + + 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( + "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( + "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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete 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( + "GET", + f"attributes/name/{attribute_name}", + AttributeGetByNameResponse, + params=params if params else None, + ) + + # ============= Attribute Group Endpoints ============= + + 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( + "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 attribute group + """ + return self._make_request( + "POST", + "attribute/groups", + SingleAttributeGroupResponse, + json_data=group_data, + ) + + def get_attribute_group_by_id(self, attribute_group_id: int, lang: Optional[str] = None) -> SingleAttributeGroupResponse: + """ + Get an attribute group by ID. + + Args: + attribute_group_id: ID of the attribute group + lang: Language code (optional) + + Returns: + SingleAttributeGroupResponse with attribute group details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"attribute/groups/{attribute_group_id}", + SingleAttributeGroupResponse, + params=params if params else None, + ) + + def delete_attribute_group(self, attribute_group_id: int) -> None: + """ + Delete an attribute group by ID. + + Args: + attribute_group_id: ID of the attribute group to delete + """ + url = urljoin(self.base_url, f"/rest/attribute/groups/{attribute_group_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") + + def get_attribute_group_by_name(self, attribute_group_name: str, lang: Optional[str] = None) -> SingleAttributeGroupResponse: + """ + Get an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group + lang: Language code (optional) + + Returns: + SingleAttributeGroupResponse with attribute group details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"attribute/groups/name/{attribute_group_name}", + SingleAttributeGroupResponse, + params=params if params else None, + ) + + def update_attribute_group_by_name(self, attribute_group_name: str, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse: + """ + Update an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group to update + group_data: Updated attribute group data + + Returns: + SingleAttributeGroupResponse with updated attribute group + """ + return self._make_request( + "PATCH", + f"attribute/groups/name/{attribute_group_name}", + SingleAttributeGroupResponse, + json_data=group_data, + ) + + def delete_attribute_group_by_name(self, attribute_group_name: str) -> None: + """ + Delete an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group to delete + """ + url = urljoin(self.base_url, f"/rest/attribute/groups/name/{attribute_group_name}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") + + def attribute_group_add_operation(self, operation_data: Dict[str, Any]) -> SingleAttributeGroupResponse: + """ + Perform an add operation on an attribute group. + + Args: + operation_data: Operation details for adding to attribute group + + Returns: + SingleAttributeGroupResponse with updated attribute group + """ + return self._make_request( + "POST", + "attribute/groups/operations/add", + SingleAttributeGroupResponse, + json_data=operation_data, + ) + + # ============= Text Endpoints ============= + + def get_all_texts( + self, lang: Optional[str] = None, page: int = 1, limit: int = 10, + include_content: bool = False, include_attributes: bool = False + ) -> TextListResponse: + """ + Get all texts with optional content and attributes. + + Args: + lang: Language code (optional) + page: Page number (default: 1) + limit: Number of texts per page (default: 10) + include_content: Include text content (default: False) + include_attributes: Include text attributes (default: False) + + Returns: + TextListResponse with paginated list of texts + """ + params: Dict[str, Any] = {"page": page, "limit": limit} + if lang: + params["lang"] = lang + if include_content: + params["includeContent"] = "true" + if include_attributes: + params["includeAttributes"] = "true" + + return self._make_request( + "GET", + "text", + TextListResponse, + params=params, + ) + + def create_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: + """ + Create a new text. + + Args: + text_data: Text data + + Returns: + SingleTextResponse with created text + """ + return self._make_request( + "POST", + "text", + SingleTextResponse, + json_data=text_data, + ) + + def update_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: + """ + Update a text. + + Args: + text_data: Updated text data (must include 'id') + + Returns: + SingleTextResponse with updated text + """ + return self._make_request( + "PATCH", + "text", + SingleTextResponse, + json_data=text_data, + ) + + def create_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkCreateResponse: + """ + Create multiple texts in bulk. + + Args: + texts_list: List of text data + + Returns: + TextBulkCreateResponse with created texts + """ + return self._make_request( + "POST", + "text/bulk", + TextBulkCreateResponse, + json_data=texts_list, + ) + + def update_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkUpdateResponse: + """ + Update multiple texts in bulk. + + Args: + texts_list: List of text data to update (each must include 'id') + + Returns: + TextBulkUpdateResponse with updated texts + """ + return self._make_request( + "PATCH", + "text/bulk", + TextBulkUpdateResponse, + json_data=texts_list, + ) + + def get_text_by_id(self, text_id: int, lang: Optional[str] = None) -> SingleTextResponse: + """ + Get a text by ID. + + Args: + text_id: ID of the text + lang: Language code (optional) + + Returns: + SingleTextResponse with text details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"text/{text_id}", + SingleTextResponse, + params=params if params else None, + ) + + def delete_text(self, text_id: int) -> None: + """ + Delete a text by ID. + + Args: + text_id: ID of the text to delete + """ + url = urljoin(self.base_url, f"/rest/text/{text_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete text: {response.status_code}") + def close(self) -> None: """Close the session and clean up resources.""" self.session.close() diff --git a/elytra_client/rest_api/models.py b/elytra_client/rest_api/models.py deleted file mode 100644 index c82a370..0000000 --- a/elytra_client/rest_api/models.py +++ /dev/null @@ -1,368 +0,0 @@ -"""Models for the Lobster PIM Legacy REST API""" - -from typing import Any, Dict, List, Optional - -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") - jobDescription: Optional[str] = Field(None, description="Description of the job") - status: str = Field(..., description="Current status of the job") - nextExecutionDate: str = Field(..., description="Next scheduled execution date") - previousExecutionDate: Optional[str] = Field(None, description="Previous execution date") - protocolId: Optional[str] = Field(None, description="ID of the associated protocol") - 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 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") - - -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") - jobDescription: Optional[str] = Field(None, description="Description of the job") - status: str = Field(..., description="Status after execution") - nextExecutionDate: str = Field(..., description="Next execution date") - 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)" - ) - 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") - username: str = Field(..., description="Username for authentication") - password: str = Field(..., description="Password for authentication") - additionalReference: Optional[str] = Field( - None, description="Custom reference for external processing tracking" - ) - 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") - maxJobDurationSeconds: Optional[int] = Field( - default=43200, description="Max duration in seconds (default 12 hours)" - ) - - -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") - - -# ============= Media and Hierarchy Models ============= - - -class PaginationLinks(BaseModel): - """Pagination links for list responses""" - - self: str = Field(..., description="Link to current page") - next: Optional[str] = Field(None, description="Link to next page") - previous: Optional[str] = Field(None, description="Link to previous page") - first: str = Field(..., description="Link to first page") - last: str = Field(..., description="Link to last page") - - -class AttributeResponse(BaseModel): - """Attribute value associated with an object""" - - id: int = Field(..., description="The ID of the attribute value") - attributeId: int = Field(..., description="The ID of the attribute definition") - attributeName: str = Field(..., description="The independent name of the attribute") - attributeType: str = Field( - ..., description="The category type of the attribute (normal, meta, internal)" - ) - type: str = Field(..., description="The type of the attribute") - value: str = Field(..., description="The value of the attribute") - parentId: Optional[int] = Field(None, description="The ID of the parent object") - autoSync: str = Field(..., description="The auto sync mode") - languageCode: Optional[str] = Field(None, description="The language code of the attribute") - modifiedAt: Optional[str] = Field( - None, description="The date and time the attribute was modified" - ) - modifiedBy: Optional[int] = Field( - None, description="The ID of the user who modified the attribute" - ) - inherited: bool = Field(False, description="Whether the attribute is inherited") - - -class MediaFileResponse(BaseModel): - """Media file metadata""" - - id: int = Field(..., description="The ID of the media file") - clientId: int = Field(..., description="The ID of the client") - mediaId: int = Field(..., description="The ID of the media descriptor") - languageCode: str = Field(..., description="The language code for this media file") - mimeType: str = Field(..., description="The MIME type of the media file") - sourceMimeType: str = Field(..., description="The original MIME type before any conversion") - mamSystem: str = Field(..., description="The Media Asset Management system name") - mamId1: Optional[str] = Field(None, description="MAM system identifier 1") - mamId2: Optional[str] = Field(None, description="MAM system identifier 2") - mamId3: Optional[str] = Field(None, description="MAM system identifier 3") - mamId4: Optional[str] = Field(None, description="MAM system identifier 4") - contentLength: int = Field(..., description="The size of the media file in bytes") - updateCount: int = Field( - ..., description="The number of times this media file has been updated" - ) - changedAt: Optional[str] = Field( - None, description="The date and time the media file was last changed" - ) - changedBy: Optional[int] = Field( - None, description="The ID of the user who last changed the media file" - ) - - -class SingleMediaResponse(BaseModel): - """Complete media descriptor""" - - id: int = Field(..., description="The ID of the media content") - name: str = Field(..., description="The name of the media") - treeId: int = Field(..., description="The ID of the tree this media belongs to") - clientId: int = Field(..., description="The ID of the client") - attributeGroupId: int = Field(..., description="The ID of the media default attribute group") - pictureTypeId: Optional[int] = Field(None, description="The ID of the picture type") - originalId: Optional[int] = Field( - None, description="The ID of the original media if this is a copy" - ) - objectStatus: Optional[str] = Field( - None, description="The status of the object (original, copy)" - ) - userObjectStatus: Optional[int] = Field(None, description="User-defined object status ID") - createdAt: Optional[str] = Field(None, description="The date and time the media was created") - createdBy: Optional[int] = Field(None, description="The ID of the user who created the media") - modifiedAt: Optional[str] = Field( - None, description="The date and time the media was last modified" - ) - modifiedBy: Optional[int] = Field( - None, description="The ID of the user who last modified the media" - ) - files: List[MediaFileResponse] = Field( - default_factory=list, description="The files associated with this media" - ) - attributes: List[AttributeResponse] = Field( - default_factory=list, description="The attribute values of the media" - ) - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the media name by language code" - ) - - -class SingleNewMediaRequestBody(BaseModel): - """Request body for creating a new media descriptor""" - - name: str = Field(..., description="Name of the media item") - attributeGroupId: Optional[int] = Field( - None, description="The ID of the media default attribute group" - ) - pictureTypeId: Optional[int] = Field(None, description="The ID of the picture type") - treeId: Optional[int] = Field(None, description="The ID of the tree this media belongs to") - originalId: Optional[int] = Field( - None, description="If this is a copy, the ID of the original media" - ) - objectStatus: Optional[str] = Field(None, description="The status of the object") - userObjectStatus: Optional[int] = Field(None, description="Custom user object status ID") - attributes: Optional[List[Dict[str, Any]]] = Field(None, description="List of media attributes") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the media name" - ) - - -class SingleUpdateMediaRequestBody(BaseModel): - """Request body for updating a media descriptor""" - - id: int = Field(..., description="The ID of the media descriptor to update") - name: Optional[str] = Field(None, description="Name of the media item") - attributeGroupId: Optional[int] = Field( - None, description="The ID of the media default attribute group" - ) - pictureTypeId: Optional[int] = Field(None, description="The ID of the picture type") - treeId: Optional[int] = Field(None, description="The ID of the tree this media belongs to") - originalId: Optional[int] = Field( - None, description="If this is a copy, the ID of the original media" - ) - objectStatus: Optional[str] = Field(None, description="The status of the object") - userObjectStatus: Optional[int] = Field(None, description="Custom user object status ID") - attributes: Optional[List[Dict[str, Any]]] = Field(None, description="List of media attributes") - translations: Optional[Dict[str, str]] = Field( - None, description="Translations of the media name" - ) - - -class MediaListResponse(BaseModel): - """Paginated list of media descriptors""" - - items: List[SingleMediaResponse] = Field(..., description="List of media items") - total: int = Field(..., description="The total number of media items") - page: int = Field(..., description="The current page number") - limit: int = Field(..., description="The number of media items per page") - links: PaginationLinks = Field(..., description="Pagination links") - - -class MediaBulkCreateResponse(BaseModel): - """Response from bulk media creation""" - - items: List[SingleMediaResponse] = Field(..., description="List of created media items") - totalItemsCreated: int = Field(..., description="The total number of media items created") - - -class MediaBulkUpdateResponse(BaseModel): - """Response from bulk media update""" - - items: List[SingleMediaResponse] = Field(..., description="List of updated media items") - totalItemsUpdated: int = Field(..., description="The total number of media items updated") - - -# ============= Hierarchy Models ============= - - -class HierarchyNode(BaseModel): - """A node in a hierarchy tree""" - - id: int = Field(..., description="The ID of the node") - name: str = Field(..., description="The name of the node") - type: str = Field(..., description="The type of node (product, variant, media, text, etc.)") - children: List["HierarchyNode"] = Field( - default_factory=list, description="The immediate children of the node" - ) - - -HierarchyNode.model_rebuild() - - -class ProductHierarchyResponse(HierarchyNode): - """Hierarchical product structure""" - - pass - - -class ProductGroupHierarchyResponse(BaseModel): - """Hierarchical product group structure""" - - id: int = Field(..., description="The ID of the node") - name: str = Field(..., description="The name of the node") - type: str = Field( - ..., description="The type of node (product-group, product, variant, media, text)" - ) - children: List["ProductGroupHierarchyResponse"] = Field( - default_factory=list, description="The immediate children" - ) - - -ProductGroupHierarchyResponse.model_rebuild() - - -class TreeGroupHierarchyResponse(BaseModel): - """Hierarchical tree group structure""" - - id: int = Field(..., description="The ID of the node") - name: str = Field(..., description="The name of the node") - type: str = Field(..., description="The type of node (tree-group, product-group)") - children: List["TreeGroupHierarchyResponse"] = Field( - default_factory=list, description="The immediate children" - ) - - -TreeGroupHierarchyResponse.model_rebuild() - - -class AttributeGroupHierarchyResponse(BaseModel): - """Hierarchical attribute group structure""" - - id: int = Field(..., description="The ID of the node") - name: str = Field(..., description="The name of the node") - type: str = Field(..., description="The type of node") - children: List["AttributeGroupHierarchyResponse"] = Field( - default_factory=list, description="The immediate children" - ) - - -AttributeGroupHierarchyResponse.model_rebuild() diff --git a/elytra_client/rest_api/models/__init__.py b/elytra_client/rest_api/models/__init__.py new file mode 100644 index 0000000..98ef30a --- /dev/null +++ b/elytra_client/rest_api/models/__init__.py @@ -0,0 +1,195 @@ +"""Models package for Lobster PIM Legacy REST API""" + +# Shared models +from .shared import AttributeResponse, ErrorResponse, PaginationLinks + +# Hierarchy models +from .hierarchy import HierarchyNode + +# Job models +from .jobs import ( + JobControlRequest, + JobControlResponse, + JobDetailInfo, + JobExecutionResponse, + JobInfo, + JobOverviewResponse, +) + +# Protocol models +from .protocols import ( + ProtocolCategoryInfo, + ProtocolCategoryListResponse, + ProtocolEntry, + ProtocolInfo, + ProtocolListResponse, +) + +# Media models +from .media import ( + MediaBulkCreateResponse, + MediaBulkUpdateResponse, + MediaFileResponse, + MediaListResponse, + SingleMediaResponse, + SingleNewMediaRequestBody, + SingleUpdateMediaRequestBody, +) + +# Product models +from .products import ( + ProductAttributeResponse, + ProductBulkCreateResponse, + ProductBulkUpdateResponse, + ProductHierarchyNode, + ProductHierarchyResponse, + ProductListResponse, + ProductOperationRequestBody, + SingleNewProductRequestBody, + SingleProductResponse, + SingleUpdateProductRequestBody, +) + +# Product group models +from .product_groups import ( + ProductGroupBulkCreateResponse, + ProductGroupBulkUpdateResponse, + ProductGroupHierarchyNode, + ProductGroupHierarchyResponse, + ProductGroupListResponse, + SingleNewProductGroupRequestBody, + SingleProductGroupResponse, + SingleUpdateProductGroupRequestBody, +) + +# Tree group models +from .tree_groups import ( + SingleNewTreeGroupRequestBody, + SingleTreeGroupResponse, + SingleUpdateTreeGroupRequestBody, + TreeGroupBulkCreateResponse, + TreeGroupBulkUpdateResponse, + TreeGroupHierarchyNode, + TreeGroupHierarchyResponse, + TreeGroupListResponse, +) + +# Attribute models +from .attributes import ( + AttributeBulkCreateResponse, + AttributeBulkUpdateResponse, + AttributeGetByNameResponse, + AttributeListResponse, + SimpleAttributeResponse, + SingleNewAttributeRequestBody, + SingleUpdateAttributeRequestBody, +) + +# Attribute group models +from .attribute_groups import ( + AttributeGroupBulkCreateResponse, + AttributeGroupHierarchyNode, + AttributeGroupHierarchyResponse, + AttributeGroupListResponse, + AttributeGroupValidFor, + SingleAttributeGroupResponse, + SingleNewAttributeGroupRequestBody, + SingleUpdateAttributeGroupRequestBody, +) + +# Text models +from .text import ( + SingleNewTextRequestBody, + SingleTextResponse, + SingleUpdateTextRequestBody, + TextBulkCreateResponse, + TextBulkUpdateResponse, + TextContentRequestBody, + TextContentResponse, + TextListResponse, +) + +__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 new file mode 100644 index 0000000..6bbcdc5 --- /dev/null +++ b/elytra_client/rest_api/models/attribute_groups.py @@ -0,0 +1,123 @@ +"""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 new file mode 100644 index 0000000..addb6b7 --- /dev/null +++ b/elytra_client/rest_api/models/attributes.py @@ -0,0 +1,75 @@ +"""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 new file mode 100644 index 0000000..017f386 --- /dev/null +++ b/elytra_client/rest_api/models/hierarchy.py @@ -0,0 +1,19 @@ +"""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/jobs.py b/elytra_client/rest_api/models/jobs.py new file mode 100644 index 0000000..9bf0923 --- /dev/null +++ b/elytra_client/rest_api/models/jobs.py @@ -0,0 +1,84 @@ +"""Job management models""" + +from typing import Any, Dict, List, Optional + +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") + jobDescription: Optional[str] = Field(None, description="Description of the job") + status: str = Field(..., description="Current status of the job") + nextExecutionDate: str = Field(..., description="Next scheduled execution date") + previousExecutionDate: Optional[str] = Field(None, description="Previous execution date") + protocolId: Optional[str] = Field(None, description="ID of the associated protocol") + 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 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") + + +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") + jobDescription: Optional[str] = Field(None, description="Description of the job") + status: str = Field(..., description="Status after execution") + nextExecutionDate: str = Field(..., description="Next execution date") + 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)" + ) + 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") + username: str = Field(..., description="Username for authentication") + password: str = Field(..., description="Password for authentication") + additionalReference: Optional[str] = Field( + None, description="Custom reference for external processing tracking" + ) + 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") + maxJobDurationSeconds: Optional[int] = Field( + default=43200, description="Max duration in seconds (default 12 hours)" + ) + + +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") diff --git a/elytra_client/rest_api/models/media.py b/elytra_client/rest_api/models/media.py new file mode 100644 index 0000000..c19b937 --- /dev/null +++ b/elytra_client/rest_api/models/media.py @@ -0,0 +1,133 @@ +"""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 new file mode 100644 index 0000000..4631ac3 --- /dev/null +++ b/elytra_client/rest_api/models/product_groups.py @@ -0,0 +1,105 @@ +"""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 new file mode 100644 index 0000000..2b3bf22 --- /dev/null +++ b/elytra_client/rest_api/models/products.py @@ -0,0 +1,120 @@ +"""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 new file mode 100644 index 0000000..b46d31c --- /dev/null +++ b/elytra_client/rest_api/models/protocols.py @@ -0,0 +1,52 @@ +"""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 new file mode 100644 index 0000000..e4911db --- /dev/null +++ b/elytra_client/rest_api/models/shared.py @@ -0,0 +1,46 @@ +"""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 new file mode 100644 index 0000000..e7f7173 --- /dev/null +++ b/elytra_client/rest_api/models/text.py @@ -0,0 +1,125 @@ +"""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 new file mode 100644 index 0000000..d4040f1 --- /dev/null +++ b/elytra_client/rest_api/models/tree_groups.py @@ -0,0 +1,87 @@ +"""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/generated_methods.py b/generated_methods.py new file mode 100644 index 0000000..b97f59d --- /dev/null +++ b/generated_methods.py @@ -0,0 +1,803 @@ +```python + # ============= Product Endpoints ============= + + 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( + "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( + "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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete 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( + "POST", + "products/operation", + ProductListResponse, + json_data=operation_data, + ) + + # ============= Product Group Endpoints ============= + + 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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete product group: {response.status_code}") + + # ============= Tree Group Endpoints ============= + + 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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete tree group: {response.status_code}") + + # ============= Attribute Endpoints ============= + + 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( + "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( + "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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete 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( + "GET", + f"attributes/name/{attribute_name}", + AttributeGetByNameResponse, + params=params if params else None, + ) + + # ============= Attribute Group Endpoints ============= + + 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( + "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 attribute group + """ + return self._make_request( + "POST", + "attribute/groups", + SingleAttributeGroupResponse, + json_data=group_data, + ) + + def get_attribute_group_by_id(self, attribute_group_id: int, lang: Optional[str] = None) -> SingleAttributeGroupResponse: + """ + Get an attribute group by ID. + + Args: + attribute_group_id: ID of the attribute group + lang: Language code (optional) + + Returns: + SingleAttributeGroupResponse with attribute group details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"attribute/groups/{attribute_group_id}", + SingleAttributeGroupResponse, + params=params if params else None, + ) + + def delete_attribute_group(self, attribute_group_id: int) -> None: + """ + Delete an attribute group by ID. + + Args: + attribute_group_id: ID of the attribute group to delete + """ + url = urljoin(self.base_url, f"/rest/attribute/groups/{attribute_group_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") + + def get_attribute_group_by_name(self, attribute_group_name: str, lang: Optional[str] = None) -> SingleAttributeGroupResponse: + """ + Get an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group + lang: Language code (optional) + + Returns: + SingleAttributeGroupResponse with attribute group details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"attribute/groups/name/{attribute_group_name}", + SingleAttributeGroupResponse, + params=params if params else None, + ) + + def update_attribute_group_by_name(self, attribute_group_name: str, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse: + """ + Update an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group to update + group_data: Updated attribute group data + + Returns: + SingleAttributeGroupResponse with updated attribute group + """ + return self._make_request( + "PATCH", + f"attribute/groups/name/{attribute_group_name}", + SingleAttributeGroupResponse, + json_data=group_data, + ) + + def delete_attribute_group_by_name(self, attribute_group_name: str) -> None: + """ + Delete an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group to delete + """ + url = urljoin(self.base_url, f"/rest/attribute/groups/name/{attribute_group_name}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") + + def attribute_group_add_operation(self, operation_data: Dict[str, Any]) -> SingleAttributeGroupResponse: + """ + Perform an add operation on an attribute group. + + Args: + operation_data: Operation details for adding to attribute group + + Returns: + SingleAttributeGroupResponse with updated attribute group + """ + return self._make_request( + "POST", + "attribute/groups/operations/add", + SingleAttributeGroupResponse, + json_data=operation_data, + ) + + # ============= Text Endpoints ============= + + def get_all_texts( + self, lang: Optional[str] = None, page: int = 1, limit: int = 10, + include_content: bool = False, include_attributes: bool = False + ) -> TextListResponse: + """ + Get all texts with optional content and attributes. + + Args: + lang: Language code (optional) + page: Page number (default: 1) + limit: Number of texts per page (default: 10) + include_content: Include text content (default: False) + include_attributes: Include text attributes (default: False) + + Returns: + TextListResponse with paginated list of texts + """ + params: Dict[str, Any] = {"page": page, "limit": limit} + if lang: + params["lang"] = lang + if include_content: + params["includeContent"] = "true" + if include_attributes: + params["includeAttributes"] = "true" + + return self._make_request( + "GET", + "text", + TextListResponse, + params=params, + ) + + def create_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: + """ + Create a new text. + + Args: + text_data: Text data + + Returns: + SingleTextResponse with created text + """ + return self._make_request( + "POST", + "text", + SingleTextResponse, + json_data=text_data, + ) + + def update_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: + """ + Update a text. + + Args: + text_data: Updated text data (must include 'id') + + Returns: + SingleTextResponse with updated text + """ + return self._make_request( + "PATCH", + "text", + SingleTextResponse, + json_data=text_data, + ) + + def create_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkCreateResponse: + """ + Create multiple texts in bulk. + + Args: + texts_list: List of text data + + Returns: + TextBulkCreateResponse with created texts + """ + return self._make_request( + "POST", + "text/bulk", + TextBulkCreateResponse, + json_data=texts_list, + ) + + def update_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkUpdateResponse: + """ + Update multiple texts in bulk. + + Args: + texts_list: List of text data to update (each must include 'id') + + Returns: + TextBulkUpdateResponse with updated texts + """ + return self._make_request( + "PATCH", + "text/bulk", + TextBulkUpdateResponse, + json_data=texts_list, + ) + + def get_text_by_id(self, text_id: int, lang: Optional[str] = None) -> SingleTextResponse: + """ + Get a text by ID. + + Args: + text_id: ID of the text + lang: Language code (optional) + + Returns: + SingleTextResponse with text details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"text/{text_id}", + SingleTextResponse, + params=params if params else None, + ) + + def delete_text(self, text_id: int) -> None: + """ + Delete a text by ID. + + Args: + text_id: ID of the text to delete + """ + url = urljoin(self.base_url, f"/rest/text/{text_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete text: {response.status_code}") +``` + diff --git a/methods_to_add.py b/methods_to_add.py new file mode 100644 index 0000000..6691f27 --- /dev/null +++ b/methods_to_add.py @@ -0,0 +1,800 @@ + # ============= Product Endpoints ============= + + 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( + "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( + "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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete 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( + "POST", + "products/operation", + ProductListResponse, + json_data=operation_data, + ) + + # ============= Product Group Endpoints ============= + + 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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete product group: {response.status_code}") + + # ============= Tree Group Endpoints ============= + + 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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete tree group: {response.status_code}") + + # ============= Attribute Endpoints ============= + + 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( + "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( + "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( + "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( + "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( + "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( + "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}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete 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( + "GET", + f"attributes/name/{attribute_name}", + AttributeGetByNameResponse, + params=params if params else None, + ) + + # ============= Attribute Group Endpoints ============= + + 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( + "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 attribute group + """ + return self._make_request( + "POST", + "attribute/groups", + SingleAttributeGroupResponse, + json_data=group_data, + ) + + def get_attribute_group_by_id(self, attribute_group_id: int, lang: Optional[str] = None) -> SingleAttributeGroupResponse: + """ + Get an attribute group by ID. + + Args: + attribute_group_id: ID of the attribute group + lang: Language code (optional) + + Returns: + SingleAttributeGroupResponse with attribute group details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"attribute/groups/{attribute_group_id}", + SingleAttributeGroupResponse, + params=params if params else None, + ) + + def delete_attribute_group(self, attribute_group_id: int) -> None: + """ + Delete an attribute group by ID. + + Args: + attribute_group_id: ID of the attribute group to delete + """ + url = urljoin(self.base_url, f"/rest/attribute/groups/{attribute_group_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") + + def get_attribute_group_by_name(self, attribute_group_name: str, lang: Optional[str] = None) -> SingleAttributeGroupResponse: + """ + Get an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group + lang: Language code (optional) + + Returns: + SingleAttributeGroupResponse with attribute group details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"attribute/groups/name/{attribute_group_name}", + SingleAttributeGroupResponse, + params=params if params else None, + ) + + def update_attribute_group_by_name(self, attribute_group_name: str, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse: + """ + Update an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group to update + group_data: Updated attribute group data + + Returns: + SingleAttributeGroupResponse with updated attribute group + """ + return self._make_request( + "PATCH", + f"attribute/groups/name/{attribute_group_name}", + SingleAttributeGroupResponse, + json_data=group_data, + ) + + def delete_attribute_group_by_name(self, attribute_group_name: str) -> None: + """ + Delete an attribute group by name. + + Args: + attribute_group_name: Name of the attribute group to delete + """ + url = urljoin(self.base_url, f"/rest/attribute/groups/name/{attribute_group_name}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}") + + def attribute_group_add_operation(self, operation_data: Dict[str, Any]) -> SingleAttributeGroupResponse: + """ + Perform an add operation on an attribute group. + + Args: + operation_data: Operation details for adding to attribute group + + Returns: + SingleAttributeGroupResponse with updated attribute group + """ + return self._make_request( + "POST", + "attribute/groups/operations/add", + SingleAttributeGroupResponse, + json_data=operation_data, + ) + + # ============= Text Endpoints ============= + + def get_all_texts( + self, lang: Optional[str] = None, page: int = 1, limit: int = 10, + include_content: bool = False, include_attributes: bool = False + ) -> TextListResponse: + """ + Get all texts with optional content and attributes. + + Args: + lang: Language code (optional) + page: Page number (default: 1) + limit: Number of texts per page (default: 10) + include_content: Include text content (default: False) + include_attributes: Include text attributes (default: False) + + Returns: + TextListResponse with paginated list of texts + """ + params: Dict[str, Any] = {"page": page, "limit": limit} + if lang: + params["lang"] = lang + if include_content: + params["includeContent"] = "true" + if include_attributes: + params["includeAttributes"] = "true" + + return self._make_request( + "GET", + "text", + TextListResponse, + params=params, + ) + + def create_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: + """ + Create a new text. + + Args: + text_data: Text data + + Returns: + SingleTextResponse with created text + """ + return self._make_request( + "POST", + "text", + SingleTextResponse, + json_data=text_data, + ) + + def update_text(self, text_data: Dict[str, Any]) -> SingleTextResponse: + """ + Update a text. + + Args: + text_data: Updated text data (must include 'id') + + Returns: + SingleTextResponse with updated text + """ + return self._make_request( + "PATCH", + "text", + SingleTextResponse, + json_data=text_data, + ) + + def create_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkCreateResponse: + """ + Create multiple texts in bulk. + + Args: + texts_list: List of text data + + Returns: + TextBulkCreateResponse with created texts + """ + return self._make_request( + "POST", + "text/bulk", + TextBulkCreateResponse, + json_data=texts_list, + ) + + def update_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkUpdateResponse: + """ + Update multiple texts in bulk. + + Args: + texts_list: List of text data to update (each must include 'id') + + Returns: + TextBulkUpdateResponse with updated texts + """ + return self._make_request( + "PATCH", + "text/bulk", + TextBulkUpdateResponse, + json_data=texts_list, + ) + + def get_text_by_id(self, text_id: int, lang: Optional[str] = None) -> SingleTextResponse: + """ + Get a text by ID. + + Args: + text_id: ID of the text + lang: Language code (optional) + + Returns: + SingleTextResponse with text details + """ + params: Dict[str, Any] = {} + if lang: + params["lang"] = lang + + return self._make_request( + "GET", + f"text/{text_id}", + SingleTextResponse, + params=params if params else None, + ) + + def delete_text(self, text_id: int) -> None: + """ + Delete a text by ID. + + Args: + text_id: ID of the text to delete + """ + url = urljoin(self.base_url, f"/rest/text/{text_id}") + params = None + if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD: + params = self.auth.get_url_parameters() + + response = self.session.delete(url, params=params, timeout=self.timeout) + if response.status_code >= 400: + raise ElytraAPIError(f"Failed to delete text: {response.status_code}") \ No newline at end of file