diff --git a/add_methods.py b/add_methods.py deleted file mode 100644 index 2847fdd..0000000 --- a/add_methods.py +++ /dev/null @@ -1,36 +0,0 @@ -#!/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/generated_methods.py b/generated_methods.py deleted file mode 100644 index b97f59d..0000000 --- a/generated_methods.py +++ /dev/null @@ -1,803 +0,0 @@ -```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 deleted file mode 100644 index 6691f27..0000000 --- a/methods_to_add.py +++ /dev/null @@ -1,800 +0,0 @@ - # ============= 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