From 9eb1f4a64108c8f2e217b5940c61d02438bb24ff Mon Sep 17 00:00:00 2001 From: claudi Date: Wed, 25 Mar 2026 10:10:03 +0100 Subject: [PATCH] feat: Add attribute and attribute group management endpoints with Pydantic validation --- elytra_client/client.py | 532 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 532 insertions(+) diff --git a/elytra_client/client.py b/elytra_client/client.py index 8b7cc4c..6f2bd43 100644 --- a/elytra_client/client.py +++ b/elytra_client/client.py @@ -13,8 +13,15 @@ from .exceptions import ( ElytraValidationError, ) from .models import ( + AttributeGroupHierarchyResponse, MediaFileResponse, + ProductGroupHierarchyResponse, + ProductHierarchyResponse, + SingleAttributeGroupResponse, + SingleAttributeResponse, SingleMediaResponse, + SingleNewAttributeGroupRequestBody, + SingleNewAttributeRequestBody, SingleNewMediaRequestBody, SingleNewProductGroupRequestBody, SingleNewProductRequestBody, @@ -24,6 +31,8 @@ from .models import ( SingleProductResponse, SingleTextResponse, SingleTreeGroupResponse, + SingleUpdateAttributeGroupRequestBody, + SingleUpdateAttributeRequestBody, SingleUpdateMediaRequestBody, SingleUpdateProductGroupRequestBody, SingleUpdateProductRequestBody, @@ -795,6 +804,529 @@ class ElytraClient: """ return self._make_request("DELETE", f"/text/{text_id}") + # Attribute extended endpoints + + def create_attribute( + self, attribute_data: SingleNewAttributeRequestBody + ) -> SingleAttributeResponse: + """ + Create a new attribute definition. + + Args: + attribute_data: Attribute definition data (Pydantic model) + + Returns: + Created attribute details (Pydantic model) + """ + return cast( + SingleAttributeResponse, + self._make_request( + "POST", + "/attributes", + json_data=attribute_data, + response_model=SingleAttributeResponse, + ), + ) + + def update_attribute( + self, attribute_data: SingleUpdateAttributeRequestBody + ) -> SingleAttributeResponse: + """ + Update an existing attribute definition. + + Args: + attribute_data: Updated attribute definition data (Pydantic model) + + Returns: + Updated attribute details (Pydantic model) + """ + return cast( + SingleAttributeResponse, + self._make_request( + "PATCH", + "/attributes", + json_data=attribute_data, + response_model=SingleAttributeResponse, + ), + ) + + def delete_attribute(self, attribute_id: int) -> Dict[str, Any]: + """ + Delete an attribute definition by ID. + + Args: + attribute_id: The attribute ID + + Returns: + Deletion response + """ + return self._make_request("DELETE", f"/attributes/{attribute_id}") + + def get_attribute_by_name( + self, attribute_name: str, lang: str = "en" + ) -> SingleAttributeResponse: + """ + Get an attribute definition by name. + + Args: + attribute_name: The attribute name (independent name) + lang: Language code + + Returns: + Attribute details (Pydantic model) + """ + params = {"lang": lang} + return cast( + SingleAttributeResponse, + self._make_request( + "GET", + f"/attributes/name/{attribute_name}", + params=params, + response_model=SingleAttributeResponse, + ), + ) + + # Attribute Group endpoints + + def get_attribute_groups( + self, lang: str = "en", page: int = 1, limit: int = 10 + ) -> Dict[str, Any]: + """ + Get all attribute groups. + + Args: + lang: Language code + page: Page number + limit: Number of attribute groups per page + + Returns: + Dictionary containing attribute groups list (validated Pydantic models) and pagination info + """ + params = {"lang": lang, "page": page, "limit": limit} + response = self._make_request("GET", "/attribute/groups", params=params) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [ + SingleAttributeGroupResponse(**item) for item in response["items"] + ] + except ValidationError as e: + raise ElytraValidationError(f"Attribute group list validation failed: {e}") + + return response + + def create_attribute_group( + self, attribute_group_data: SingleNewAttributeGroupRequestBody + ) -> SingleAttributeGroupResponse: + """ + Create a new attribute group. + + Args: + attribute_group_data: Attribute group data (Pydantic model) + + Returns: + Created attribute group details (Pydantic model) + """ + return cast( + SingleAttributeGroupResponse, + self._make_request( + "POST", + "/attribute/groups", + json_data=attribute_group_data, + response_model=SingleAttributeGroupResponse, + ), + ) + + def update_attribute_group( + self, attribute_group_data: SingleUpdateAttributeGroupRequestBody + ) -> SingleAttributeGroupResponse: + """ + Update an existing attribute group. + + Args: + attribute_group_data: Updated attribute group data (Pydantic model) + + Returns: + Updated attribute group details (Pydantic model) + """ + return cast( + SingleAttributeGroupResponse, + self._make_request( + "PATCH", + "/attribute/groups", + json_data=attribute_group_data, + response_model=SingleAttributeGroupResponse, + ), + ) + + def get_attribute_group_by_id( + self, attribute_group_id: int, lang: str = "en" + ) -> SingleAttributeGroupResponse: + """ + Get an attribute group by ID. + + Args: + attribute_group_id: The attribute group ID + lang: Language code + + Returns: + Attribute group details (Pydantic model) + """ + params = {"lang": lang} + return cast( + SingleAttributeGroupResponse, + self._make_request( + "GET", + f"/attribute/groups/id/{attribute_group_id}", + params=params, + response_model=SingleAttributeGroupResponse, + ), + ) + + def get_attribute_group_by_name( + self, attribute_group_name: str, lang: str = "en" + ) -> SingleAttributeGroupResponse: + """ + Get an attribute group by name. + + Args: + attribute_group_name: The attribute group name + lang: Language code + + Returns: + Attribute group details (Pydantic model) + """ + params = {"lang": lang} + return cast( + SingleAttributeGroupResponse, + self._make_request( + "GET", + f"/attribute/groups/name/{attribute_group_name}", + params=params, + response_model=SingleAttributeGroupResponse, + ), + ) + + def add_attributes_to_group( + self, attribute_group_id: int, attribute_ids: List[int] + ) -> Dict[str, Any]: + """ + Add attributes to an attribute group. + + Args: + attribute_group_id: The attribute group ID + attribute_ids: List of attribute IDs to add + + Returns: + Operation response + """ + data = { + "attributeGroupId": attribute_group_id, + "attributeIds": attribute_ids, + } + return self._make_request( + "POST", + "/attribute/groups/operations/add", + json_data=data, + ) + + def get_attribute_group_hierarchy( + self, attribute_group_id: int + ) -> AttributeGroupHierarchyResponse: + """ + Get the hierarchy of an attribute group. + + Args: + attribute_group_id: The attribute group ID + + Returns: + Attribute group hierarchy (Pydantic model) + """ + return cast( + AttributeGroupHierarchyResponse, + self._make_request( + "GET", + f"/attribute/groups/hierarchy/{attribute_group_id}", + response_model=AttributeGroupHierarchyResponse, + ), + ) + + # Hierarchy endpoints + + def get_product_hierarchy(self, product_id: int, depth: int = 10) -> ProductHierarchyResponse: + """ + Get the hierarchy of a product. + + Args: + product_id: The product ID + depth: The depth of the hierarchy (default: 10) + + Returns: + Product hierarchy (Pydantic model) + """ + params = {"depth": depth} + return cast( + ProductHierarchyResponse, + self._make_request( + "GET", + f"/products/{product_id}/hierarchy", + params=params, + response_model=ProductHierarchyResponse, + ), + ) + + def get_product_group_hierarchy( + self, group_id: int, depth: int = 10 + ) -> ProductGroupHierarchyResponse: + """ + Get the hierarchy of a product group. + + Args: + group_id: The product group ID + depth: The depth of the hierarchy (default: 10) + + Returns: + Product group hierarchy (Pydantic model) + """ + params = {"depth": depth} + return cast( + ProductGroupHierarchyResponse, + self._make_request( + "GET", + f"/groups/{group_id}/hierarchy", + params=params, + response_model=ProductGroupHierarchyResponse, + ), + ) + + # Product operations + + def perform_product_operation( + self, + operation: str, + product_id: int, + parent_id: int, + ) -> SingleProductResponse: + """ + Perform an operation on a product. + + Available operations: + - **copy**: Copy the product with all children to a new parent. Requires a product group as parent. + - **move**: Move the product to a new parent. Accepts product or product group as parent. + - **link**: Link the product to a new parent. Accepts product or product group as parent. + - **copy-structure**: Copy the group structure but link products. Requires a product group as parent. + + Args: + operation: The operation to perform (copy, move, link, copy-structure) + product_id: The ID of the product to perform the operation on + parent_id: The ID of the destination parent + + Returns: + Updated product details (Pydantic model) + + Raises: + ElytraValidationError: If operation requires specific parent type not provided + """ + data = { + "operation": operation, + "productId": product_id, + "parentId": parent_id, + } + return cast( + SingleProductResponse, + self._make_request( + "POST", + "/products/operation", + json_data=data, + response_model=SingleProductResponse, + ), + ) + + # Bulk operations - Products + + def create_products_bulk( + self, products_data: List[SingleNewProductRequestBody] + ) -> Dict[str, Any]: + """ + Create multiple products in bulk. + + Args: + products_data: List of product data (Pydantic models) + + Returns: + Response with created items and total count + """ + response = self._make_request("POST", "/products/bulk", json_data=products_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleProductResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk products creation validation failed: {e}") + + return response + + def update_products_bulk( + self, products_data: List[SingleUpdateProductRequestBody] + ) -> Dict[str, Any]: + """ + Update multiple products in bulk. + + Args: + products_data: List of updated product data (Pydantic models) + + Returns: + Response with updated items and total count + """ + response = self._make_request("PATCH", "/products/bulk", json_data=products_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleProductResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk products update validation failed: {e}") + + return response + + # Bulk operations - Attributes + + def create_attributes_bulk( + self, attributes_data: List[SingleNewAttributeRequestBody] + ) -> Dict[str, Any]: + """ + Create multiple attribute definitions in bulk. + + Args: + attributes_data: List of attribute definition data (Pydantic models) + + Returns: + Response with created items and total count + """ + response = self._make_request("POST", "/attributes/bulk", json_data=attributes_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleAttributeResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk attributes creation validation failed: {e}") + + return response + + def update_attributes_bulk( + self, attributes_data: List[SingleUpdateAttributeRequestBody] + ) -> Dict[str, Any]: + """ + Update multiple attribute definitions in bulk. + + Args: + attributes_data: List of updated attribute definition data (Pydantic models) + + Returns: + Response with updated items and total count + """ + response = self._make_request("PATCH", "/attributes/bulk", json_data=attributes_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleAttributeResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk attributes update validation failed: {e}") + + return response + + # Bulk operations - Media + + def create_media_bulk(self, media_data: List[SingleNewMediaRequestBody]) -> Dict[str, Any]: + """ + Create multiple media items in bulk. + + Args: + media_data: List of media data (Pydantic models) + + Returns: + Response with created items and total count + """ + response = self._make_request("POST", "/media/bulk", json_data=media_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleMediaResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk media creation validation failed: {e}") + + return response + + def update_media_bulk(self, media_data: List[SingleUpdateMediaRequestBody]) -> Dict[str, Any]: + """ + Update multiple media items in bulk. + + Args: + media_data: List of updated media data (Pydantic models) + + Returns: + Response with updated items and total count + """ + response = self._make_request("PATCH", "/media/bulk", json_data=media_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleMediaResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk media update validation failed: {e}") + + return response + + # Bulk operations - Text + + def create_texts_bulk(self, texts_data: List[SingleNewTextRequestBody]) -> Dict[str, Any]: + """ + Create multiple text items in bulk. + + Args: + texts_data: List of text data (Pydantic models) + + Returns: + Response with created items and total count + """ + response = self._make_request("POST", "/text/bulk", json_data=texts_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleTextResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk text creation validation failed: {e}") + + return response + + def update_texts_bulk(self, texts_data: List[SingleUpdateTextRequestBody]) -> Dict[str, Any]: + """ + Update multiple text items in bulk. + + Args: + texts_data: List of updated text data (Pydantic models) + + Returns: + Response with updated items and total count + """ + response = self._make_request("PATCH", "/text/bulk", json_data=texts_data) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleTextResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Bulk text update validation failed: {e}") + + return response + # Health check def health_check(self) -> Dict[str, Any]: