From 798c76376526773731790d7d6e384ab9778da0ea Mon Sep 17 00:00:00 2001 From: claudi Date: Wed, 25 Mar 2026 10:04:24 +0100 Subject: [PATCH] feat: Add tree and media endpoints with Pydantic validation --- elytra_client/client.py | 446 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 418 insertions(+), 28 deletions(-) diff --git a/elytra_client/client.py b/elytra_client/client.py index 25d1213..8b7cc4c 100644 --- a/elytra_client/client.py +++ b/elytra_client/client.py @@ -1,8 +1,9 @@ """Main Elytra PIM API Client - Fully Pydantic Driven""" -import requests -from typing import Optional, Dict, Any, List, TypeVar, Type, cast +from typing import Any, Dict, List, Optional, Type, TypeVar, cast from urllib.parse import urljoin + +import requests from pydantic import BaseModel, ValidationError from .exceptions import ( @@ -12,15 +13,26 @@ from .exceptions import ( ElytraValidationError, ) from .models import ( - SingleProductResponse, - SingleProductGroupResponse, - SingleNewProductRequestBody, - SingleUpdateProductRequestBody, + MediaFileResponse, + SingleMediaResponse, + SingleNewMediaRequestBody, SingleNewProductGroupRequestBody, + SingleNewProductRequestBody, + SingleNewTextRequestBody, + SingleNewTreeGroupRequestBody, + SingleProductGroupResponse, + SingleProductResponse, + SingleTextResponse, + SingleTreeGroupResponse, + SingleUpdateMediaRequestBody, SingleUpdateProductGroupRequestBody, + SingleUpdateProductRequestBody, + SingleUpdateTextRequestBody, + SingleUpdateTreeGroupRequestBody, + TreeGroupHierarchyResponse, ) -T = TypeVar('T', bound=BaseModel) +T = TypeVar("T", bound=BaseModel) class ElytraClient: @@ -87,7 +99,7 @@ class ElytraClient: ElytraAPIError: For other API errors """ url = urljoin(self.base_url, endpoint) - + # Convert Pydantic model to dict if needed json_payload = None if json_data is not None: @@ -106,7 +118,7 @@ class ElytraClient: ) response.raise_for_status() data = response.json() - + # Validate response with Pydantic model if provided if response_model is not None: try: @@ -115,11 +127,11 @@ class ElytraClient: raise ElytraValidationError( f"Response validation failed: {e}", response.status_code ) - + return data except requests.exceptions.HTTPError as e: self._handle_http_error(e) - raise # Re-raise after handling to satisfy type checker, + raise # Re-raise after handling to satisfy type checker, # even though _handle_http_error will raise a specific exception and this line should never be reached except requests.exceptions.RequestException as e: raise ElytraAPIError(f"Request failed: {str(e)}") @@ -169,16 +181,14 @@ class ElytraClient: params["groupId"] = group_id response = self._make_request("GET", "/products", params=params) - + # Validate items with Pydantic models if isinstance(response, dict) and "items" in response: try: - response["items"] = [ - SingleProductResponse(**item) for item in response["items"] - ] + response["items"] = [SingleProductResponse(**item) for item in response["items"]] except ValidationError as e: raise ElytraValidationError(f"Product list validation failed: {e}") - + return response def get_product(self, product_id: int, lang: str = "en") -> SingleProductResponse: @@ -221,7 +231,7 @@ class ElytraClient: """ params = {"lang": lang, "page": page, "limit": limit} response = self._make_request("GET", "/groups", params=params) - + # Validate items with Pydantic models if isinstance(response, dict) and "items" in response: try: @@ -230,7 +240,7 @@ class ElytraClient: ] except ValidationError as e: raise ElytraValidationError(f"Product group list validation failed: {e}") - + return response def get_product_group(self, group_id: int, lang: str = "en") -> SingleProductGroupResponse: @@ -257,9 +267,7 @@ class ElytraClient: # Attribute endpoints - def get_attributes( - self, lang: str = "en", page: int = 1, limit: int = 10 - ) -> Dict[str, Any]: + def get_attributes(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: """ Get all attributes. @@ -273,7 +281,7 @@ class ElytraClient: """ params = {"lang": lang, "page": page, "limit": limit} response = self._make_request("GET", "/attributes", params=params) - + return response def get_attribute(self, attribute_id: int, lang: str = "en") -> Dict[str, Any]: @@ -292,9 +300,7 @@ class ElytraClient: # Product CRUD operations with Pydantic validation - def create_product( - self, product_data: SingleNewProductRequestBody - ) -> SingleProductResponse: + def create_product(self, product_data: SingleNewProductRequestBody) -> SingleProductResponse: """ Create a new product. @@ -314,9 +320,7 @@ class ElytraClient: ), ) - def update_product( - self, product_data: SingleUpdateProductRequestBody - ) -> SingleProductResponse: + def update_product(self, product_data: SingleUpdateProductRequestBody) -> SingleProductResponse: """ Update an existing product. @@ -406,6 +410,391 @@ class ElytraClient: """ return self._make_request("DELETE", f"/groups/{group_id}") + # Tree Group endpoints + + def get_tree_groups(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: + """ + Get all tree groups. + + Args: + lang: Language code + page: Page number + limit: Number of tree groups per page + + Returns: + Dictionary containing tree groups list (validated Pydantic models) and pagination info + """ + params = {"lang": lang, "page": page, "limit": limit} + response = self._make_request("GET", "/tree/groups", params=params) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleTreeGroupResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Tree group list validation failed: {e}") + + return response + + def get_tree_group(self, tree_group_id: int, lang: str = "en") -> SingleTreeGroupResponse: + """ + Get a single tree group by ID with its children. + + Args: + tree_group_id: The tree group ID + lang: Language code + + Returns: + Tree group details with children (Pydantic model) + """ + params = {"lang": lang} + return cast( + SingleTreeGroupResponse, + self._make_request( + "GET", + f"/tree/groups/{tree_group_id}", + params=params, + response_model=SingleTreeGroupResponse, + ), + ) + + def create_tree_group( + self, tree_group_data: SingleNewTreeGroupRequestBody + ) -> SingleTreeGroupResponse: + """ + Create a new tree group. + + Args: + tree_group_data: Tree group data (Pydantic model) + + Returns: + Created tree group details (Pydantic model) + """ + return cast( + SingleTreeGroupResponse, + self._make_request( + "POST", + "/tree/groups", + json_data=tree_group_data, + response_model=SingleTreeGroupResponse, + ), + ) + + def update_tree_group( + self, tree_group_data: SingleUpdateTreeGroupRequestBody + ) -> SingleTreeGroupResponse: + """ + Update an existing tree group. + + Args: + tree_group_data: Updated tree group data (Pydantic model) + + Returns: + Updated tree group details (Pydantic model) + """ + return cast( + SingleTreeGroupResponse, + self._make_request( + "PATCH", + "/tree/groups", + json_data=tree_group_data, + response_model=SingleTreeGroupResponse, + ), + ) + + def delete_tree_group(self, tree_group_id: int) -> Dict[str, Any]: + """ + Delete a tree group by ID. + + Args: + tree_group_id: The tree group ID + + Returns: + Deletion response + """ + return self._make_request("DELETE", f"/tree/groups/{tree_group_id}") + + def get_tree_group_hierarchy(self, depth: int = 10) -> TreeGroupHierarchyResponse: + """ + Get the hierarchy of tree groups. + + Args: + depth: The depth of the hierarchy (default: 10) + + Returns: + Tree group hierarchy (Pydantic model) + """ + params = {"depth": depth} + return cast( + TreeGroupHierarchyResponse, + self._make_request( + "GET", + "/tree/groups/hierarchy", + params=params, + response_model=TreeGroupHierarchyResponse, + ), + ) + + # Media endpoints + + def get_media(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: + """ + Get all media items. + + Args: + lang: Language code + page: Page number + limit: Number of media items per page + + Returns: + Dictionary containing media list (validated Pydantic models) and pagination info + """ + params = {"lang": lang, "page": page, "limit": limit} + response = self._make_request("GET", "/media", params=params) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleMediaResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Media list validation failed: {e}") + + return response + + def get_media_by_id(self, media_id: int, lang: str = "en") -> SingleMediaResponse: + """ + Get a single media item by ID. + + Args: + media_id: The media ID + lang: Language code + + Returns: + Media details (Pydantic model) + """ + params = {"lang": lang} + return cast( + SingleMediaResponse, + self._make_request( + "GET", + f"/media/{media_id}", + params=params, + response_model=SingleMediaResponse, + ), + ) + + def create_media(self, media_data: SingleNewMediaRequestBody) -> SingleMediaResponse: + """ + Create a new media item. + + Args: + media_data: Media data (Pydantic model) + + Returns: + Created media details (Pydantic model) + """ + return cast( + SingleMediaResponse, + self._make_request( + "POST", + "/media", + json_data=media_data, + response_model=SingleMediaResponse, + ), + ) + + def update_media(self, media_data: SingleUpdateMediaRequestBody) -> SingleMediaResponse: + """ + Update an existing media item. + + Args: + media_data: Updated media data (Pydantic model) + + Returns: + Updated media details (Pydantic model) + """ + return cast( + SingleMediaResponse, + self._make_request( + "PATCH", + "/media", + json_data=media_data, + response_model=SingleMediaResponse, + ), + ) + + def delete_media(self, media_id: int) -> Dict[str, Any]: + """ + Delete a media item by ID. + + Args: + media_id: The media ID + + Returns: + Deletion response + """ + return self._make_request("DELETE", f"/media/{media_id}") + + def upload_media_file( + self, + media_id: int, + file_path: str, + language_code: str = "en", + ) -> Dict[str, Any]: + """ + Upload a file for a media item. + + Args: + media_id: The media ID + file_path: Path to the file to upload + language_code: Language code for the file + + Returns: + Upload response + + Note: + This method uses multipart/form-data encoding for file upload. + The actual file upload implementation may require custom handling + outside of the standard JSON request pattern. + """ + # Note: This is a placeholder for file upload functionality + # Actual implementation would need to handle multipart/form-data + raise NotImplementedError( + "File upload requires special multipart/form-data handling. " + "Please use direct requests session or a specialized file upload method." + ) + + def download_media_file(self, media_file_id: int) -> bytes: + """ + Download a media file. + + Args: + media_file_id: The media file ID + + Returns: + File content as bytes + + Note: + This method returns raw binary data and should be handled differently + from typical JSON API responses. + """ + raise NotImplementedError( + "File download requires special binary data handling. " + "Please use direct requests session or a specialized file download method." + ) + + def delete_media_file(self, media_file_id: int) -> Dict[str, Any]: + """ + Delete a media file. + + Args: + media_file_id: The media file ID + + Returns: + Deletion response + """ + return self._make_request("DELETE", f"/media/file/{media_file_id}") + + # Text endpoints + + def get_texts(self, lang: str = "en", page: int = 1, limit: int = 10) -> Dict[str, Any]: + """ + Get all text items. + + Args: + lang: Language code + page: Page number + limit: Number of text items per page + + Returns: + Dictionary containing text list (validated Pydantic models) and pagination info + """ + params = {"lang": lang, "page": page, "limit": limit} + response = self._make_request("GET", "/text", params=params) + + # Validate items with Pydantic models + if isinstance(response, dict) and "items" in response: + try: + response["items"] = [SingleTextResponse(**item) for item in response["items"]] + except ValidationError as e: + raise ElytraValidationError(f"Text list validation failed: {e}") + + return response + + def get_text(self, text_id: int, lang: str = "en") -> SingleTextResponse: + """ + Get a single text item by ID. + + Args: + text_id: The text ID + lang: Language code + + Returns: + Text details (Pydantic model) + """ + params = {"lang": lang} + return cast( + SingleTextResponse, + self._make_request( + "GET", + f"/text/{text_id}", + params=params, + response_model=SingleTextResponse, + ), + ) + + def create_text(self, text_data: SingleNewTextRequestBody) -> SingleTextResponse: + """ + Create a new text item. + + Args: + text_data: Text data (Pydantic model) + + Returns: + Created text details (Pydantic model) + """ + return cast( + SingleTextResponse, + self._make_request( + "POST", + "/text", + json_data=text_data, + response_model=SingleTextResponse, + ), + ) + + def update_text(self, text_data: SingleUpdateTextRequestBody) -> SingleTextResponse: + """ + Update an existing text item. + + Args: + text_data: Updated text data (Pydantic model) + + Returns: + Updated text details (Pydantic model) + """ + return cast( + SingleTextResponse, + self._make_request( + "PATCH", + "/text", + json_data=text_data, + response_model=SingleTextResponse, + ), + ) + + def delete_text(self, text_id: int) -> Dict[str, Any]: + """ + Delete a text item by ID. + + Args: + text_id: The text ID + + Returns: + Deletion response + """ + return self._make_request("DELETE", f"/text/{text_id}") + # Health check def health_check(self) -> Dict[str, Any]: @@ -428,3 +817,4 @@ class ElytraClient: def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit""" self.close() + self.close()