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

Add product, product group, tree group, attribute, and text endpoints with CRUD operations

Browse source
This commit is contained in:
claudi 2026-03-24 15:35:20 +01:00
parent ef53746129
commit 68f0b76feb
17 changed files with 3639 additions and 368 deletions
Download patch file Download diff file Expand all files Collapse all files

36
add_methods.py Normal file
View file

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

836
elytra_client/rest_api/client.py
View file

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

368
elytra_client/rest_api/models.py
View file

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

195
elytra_client/rest_api/models/__init__.py Normal file
View file

@ -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",
]

123
elytra_client/rest_api/models/attribute_groups.py Normal file
View file

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

75
elytra_client/rest_api/models/attributes.py Normal file
View file

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

19
elytra_client/rest_api/models/hierarchy.py Normal file
View file

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

84
elytra_client/rest_api/models/jobs.py Normal file
View file

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

133
elytra_client/rest_api/models/media.py Normal file
View file

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

105
elytra_client/rest_api/models/product_groups.py Normal file
View file

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

120
elytra_client/rest_api/models/products.py Normal file
View file

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

52
elytra_client/rest_api/models/protocols.py Normal file
View file

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

46
elytra_client/rest_api/models/shared.py Normal file
View file

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

125
elytra_client/rest_api/models/text.py Normal file
View file

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

87
elytra_client/rest_api/models/tree_groups.py Normal file
View file

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

803
generated_methods.py Normal file
View file

@ -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}")
```

800
methods_to_add.py Normal file
View file

@ -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}")