HIM-public/elytra_client
HIM-public/elytra_client
5
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages 1 Wiki Activity Actions

feat: Add tree and media endpoints with Pydantic validation

Browse source
This commit is contained in:
claudi 2026-03-25 10:04:24 +01:00
parent abd4e2c9ca
commit 798c763765
1 changed files with 418 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

446
elytra_client/client.py
View file

@ -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()