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

cleanup

Browse source
This commit is contained in:
claudi 2026-03-25 09:08:58 +01:00
parent aee3a48687
commit abd4e2c9ca
3 changed files with 0 additions and 1639 deletions
Download patch file Download diff file Expand all files Collapse all files

36
add_methods.py
View file

@ -1,36 +0,0 @@
#!/usr/bin/env python3
"""Script to add all missing methods to the REST client"""
import re
# Read the generated methods file
with open("generated_methods.py", "r") as f:
generated_content = f.read()
# Extract the Python code from markdown backticks
match = re.search(r"```python\n(.*?)\n```", generated_content, re.DOTALL)
if not match:
print("ERROR: Could not find Python code block in generated_methods.py")
exit(1)
methods_code = match.group(1)
# Read the current client file
with open("elytra_client/rest_api/client.py", "r") as f:
client_content = f.read()
# Find the position of "def close(self)" method
close_pos = client_content.find(" def close(self) -> None:")
if close_pos == -1:
print("ERROR: Could not find 'def close' method in client.py")
exit(1)
# Insert the new methods before the close() method
new_client_content = client_content[:close_pos] + methods_code + "\n\n" + client_content[close_pos:]
# Write back to the client file
with open("elytra_client/rest_api/client.py", "w") as f:
f.write(new_client_content)
print(f"Successfully added {methods_code.count('def ')} new methods to the client")
print("Client file updated successfully")

803
generated_methods.py
View file

@ -1,803 +0,0 @@
```python
# ============= Product Endpoints =============
def get_all_products(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10, group_id: Optional[int] = None
) -> ProductListResponse:
"""
Get all products with optional group filter.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of products per page (default: 10)
group_id: Optional product group ID to filter products
Returns:
ProductListResponse with paginated list of products
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
if group_id:
params["groupId"] = group_id
return self._make_request(
"GET",
"products",
ProductListResponse,
params=params,
)
def create_product(self, product_data: Dict[str, Any]) -> SingleProductResponse:
"""
Create a new product.
Args:
product_data: Product data
Returns:
SingleProductResponse with created product
"""
return self._make_request(
"POST",
"products",
SingleProductResponse,
json_data=product_data,
)
def update_product(self, product_data: Dict[str, Any]) -> ProductListResponse:
"""
Update a product.
Args:
product_data: Updated product data (must include 'id')
Returns:
ProductListResponse with updated product info
"""
return self._make_request(
"PATCH",
"products",
ProductListResponse,
json_data=product_data,
)
def create_multiple_products(self, products_list: List[Dict[str, Any]]) -> ProductBulkCreateResponse:
"""
Create multiple products in bulk.
Args:
products_list: List of product data
Returns:
ProductBulkCreateResponse with created products
"""
return self._make_request(
"POST",
"products/bulk",
ProductBulkCreateResponse,
json_data=products_list,
)
def update_multiple_products(self, products_list: List[Dict[str, Any]]) -> ProductBulkUpdateResponse:
"""
Update multiple products in bulk.
Args:
products_list: List of product data to update (each must include 'id')
Returns:
ProductBulkUpdateResponse with updated products
"""
return self._make_request(
"PATCH",
"products/bulk",
ProductBulkUpdateResponse,
json_data=products_list,
)
def get_product_by_id(self, product_id: int, lang: Optional[str] = None) -> SingleProductResponse:
"""
Get a product by ID.
Args:
product_id: ID of the product
lang: Language code (optional)
Returns:
SingleProductResponse with product details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"products/{product_id}",
SingleProductResponse,
params=params if params else None,
)
def delete_product(self, product_id: int) -> None:
"""
Delete a product by ID.
Args:
product_id: ID of the product to delete
"""
url = urljoin(self.base_url, f"/rest/products/{product_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete product: {response.status_code}")
def product_operation(self, operation_data: Dict[str, Any]) -> ProductListResponse:
"""
Perform bulk operations on products (copy, move, link, copy-structure).
Args:
operation_data: Operation details including operation type, source, target, etc.
Returns:
ProductListResponse with affected products
"""
return self._make_request(
"POST",
"products/operation",
ProductListResponse,
json_data=operation_data,
)
# ============= Product Group Endpoints =============
def get_all_product_groups(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> ProductGroupListResponse:
"""
Get all product groups.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of groups per page (default: 10)
Returns:
ProductGroupListResponse with paginated list of groups
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"groups",
ProductGroupListResponse,
params=params,
)
def create_product_group(self, group_data: Dict[str, Any]) -> SingleProductGroupResponse:
"""
Create a new product group.
Args:
group_data: Product group data
Returns:
SingleProductGroupResponse with created group
"""
return self._make_request(
"POST",
"groups",
SingleProductGroupResponse,
json_data=group_data,
)
def update_product_group(self, group_data: Dict[str, Any]) -> ProductGroupListResponse:
"""
Update a product group.
Args:
group_data: Updated group data (must include 'id')
Returns:
ProductGroupListResponse with updated group info
"""
return self._make_request(
"PATCH",
"groups",
ProductGroupListResponse,
json_data=group_data,
)
def get_product_group_by_id(self, group_id: int, lang: Optional[str] = None) -> SingleProductGroupResponse:
"""
Get a product group by ID.
Args:
group_id: ID of the product group
lang: Language code (optional)
Returns:
SingleProductGroupResponse with group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"groups/{group_id}",
SingleProductGroupResponse,
params=params if params else None,
)
def delete_product_group(self, group_id: int) -> None:
"""
Delete a product group by ID.
Args:
group_id: ID of the product group to delete
"""
url = urljoin(self.base_url, f"/rest/groups/{group_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete product group: {response.status_code}")
# ============= Tree Group Endpoints =============
def get_all_tree_groups(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> TreeGroupListResponse:
"""
Get all tree groups.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of groups per page (default: 10)
Returns:
TreeGroupListResponse with paginated list of tree groups
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"tree/groups",
TreeGroupListResponse,
params=params,
)
def create_tree_group(self, group_data: Dict[str, Any]) -> SingleTreeGroupResponse:
"""
Create a new tree group.
Args:
group_data: Tree group data
Returns:
SingleTreeGroupResponse with created tree group
"""
return self._make_request(
"POST",
"tree/groups",
SingleTreeGroupResponse,
json_data=group_data,
)
def update_tree_group(self, group_data: Dict[str, Any]) -> SingleTreeGroupResponse:
"""
Update a tree group.
Args:
group_data: Updated tree group data (must include 'id')
Returns:
SingleTreeGroupResponse with updated tree group
"""
return self._make_request(
"PATCH",
"tree/groups",
SingleTreeGroupResponse,
json_data=group_data,
)
def get_tree_group_by_id(self, tree_group_id: int, lang: Optional[str] = None) -> SingleTreeGroupResponse:
"""
Get a tree group by ID.
Args:
tree_group_id: ID of the tree group
lang: Language code (optional)
Returns:
SingleTreeGroupResponse with tree group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"tree/groups/{tree_group_id}",
SingleTreeGroupResponse,
params=params if params else None,
)
def delete_tree_group(self, tree_group_id: int) -> None:
"""
Delete a tree group by ID.
Args:
tree_group_id: ID of the tree group to delete
"""
url = urljoin(self.base_url, f"/rest/tree/groups/{tree_group_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete tree group: {response.status_code}")
# ============= Attribute Endpoints =============
def get_all_attributes(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> AttributeListResponse:
"""
Get all attributes.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of attributes per page (default: 10)
Returns:
AttributeListResponse with paginated list of attributes
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"attributes",
AttributeListResponse,
params=params,
)
def create_attribute(self, attribute_data: Dict[str, Any]) -> SimpleAttributeResponse:
"""
Create a new attribute.
Args:
attribute_data: Attribute data
Returns:
SimpleAttributeResponse with created attribute
"""
return self._make_request(
"POST",
"attributes",
SimpleAttributeResponse,
json_data=attribute_data,
)
def update_attribute(self, attribute_data: Dict[str, Any]) -> SimpleAttributeResponse:
"""
Update an attribute.
Args:
attribute_data: Updated attribute data (must include 'id')
Returns:
SimpleAttributeResponse with updated attribute
"""
return self._make_request(
"PATCH",
"attributes",
SimpleAttributeResponse,
json_data=attribute_data,
)
def create_multiple_attributes(self, attributes_list: List[Dict[str, Any]]) -> AttributeBulkCreateResponse:
"""
Create multiple attributes in bulk.
Args:
attributes_list: List of attribute data
Returns:
AttributeBulkCreateResponse with created attributes
"""
return self._make_request(
"POST",
"attributes/bulk",
AttributeBulkCreateResponse,
json_data=attributes_list,
)
def update_multiple_attributes(self, attributes_list: List[Dict[str, Any]]) -> AttributeBulkUpdateResponse:
"""
Update multiple attributes in bulk.
Args:
attributes_list: List of attribute data to update (each must include 'id')
Returns:
AttributeBulkUpdateResponse with updated attributes
"""
return self._make_request(
"PATCH",
"attributes/bulk",
AttributeBulkUpdateResponse,
json_data=attributes_list,
)
def get_attribute_by_id(self, attribute_id: int, lang: Optional[str] = None) -> SimpleAttributeResponse:
"""
Get an attribute by ID.
Args:
attribute_id: ID of the attribute
lang: Language code (optional)
Returns:
SimpleAttributeResponse with attribute details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attributes/{attribute_id}",
SimpleAttributeResponse,
params=params if params else None,
)
def delete_attribute(self, attribute_id: int) -> None:
"""
Delete an attribute by ID.
Args:
attribute_id: ID of the attribute to delete
"""
url = urljoin(self.base_url, f"/rest/attributes/{attribute_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete attribute: {response.status_code}")
def get_attribute_by_name(self, attribute_name: str, lang: Optional[str] = None) -> AttributeGetByNameResponse:
"""
Get an attribute by name with language-dependent properties.
Args:
attribute_name: Name of the attribute
lang: Language code (optional)
Returns:
AttributeGetByNameResponse with attribute details and languageDependents
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attributes/name/{attribute_name}",
AttributeGetByNameResponse,
params=params if params else None,
)
# ============= Attribute Group Endpoints =============
def get_all_attribute_groups(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> AttributeGroupListResponse:
"""
Get all attribute groups.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of groups per page (default: 10)
Returns:
AttributeGroupListResponse with paginated list of attribute groups
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"attribute/groups",
AttributeGroupListResponse,
params=params,
)
def create_attribute_group(self, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse:
"""
Create a new attribute group.
Args:
group_data: Attribute group data
Returns:
SingleAttributeGroupResponse with created attribute group
"""
return self._make_request(
"POST",
"attribute/groups",
SingleAttributeGroupResponse,
json_data=group_data,
)
def get_attribute_group_by_id(self, attribute_group_id: int, lang: Optional[str] = None) -> SingleAttributeGroupResponse:
"""
Get an attribute group by ID.
Args:
attribute_group_id: ID of the attribute group
lang: Language code (optional)
Returns:
SingleAttributeGroupResponse with attribute group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attribute/groups/{attribute_group_id}",
SingleAttributeGroupResponse,
params=params if params else None,
)
def delete_attribute_group(self, attribute_group_id: int) -> None:
"""
Delete an attribute group by ID.
Args:
attribute_group_id: ID of the attribute group to delete
"""
url = urljoin(self.base_url, f"/rest/attribute/groups/{attribute_group_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}")
def get_attribute_group_by_name(self, attribute_group_name: str, lang: Optional[str] = None) -> SingleAttributeGroupResponse:
"""
Get an attribute group by name.
Args:
attribute_group_name: Name of the attribute group
lang: Language code (optional)
Returns:
SingleAttributeGroupResponse with attribute group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attribute/groups/name/{attribute_group_name}",
SingleAttributeGroupResponse,
params=params if params else None,
)
def update_attribute_group_by_name(self, attribute_group_name: str, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse:
"""
Update an attribute group by name.
Args:
attribute_group_name: Name of the attribute group to update
group_data: Updated attribute group data
Returns:
SingleAttributeGroupResponse with updated attribute group
"""
return self._make_request(
"PATCH",
f"attribute/groups/name/{attribute_group_name}",
SingleAttributeGroupResponse,
json_data=group_data,
)
def delete_attribute_group_by_name(self, attribute_group_name: str) -> None:
"""
Delete an attribute group by name.
Args:
attribute_group_name: Name of the attribute group to delete
"""
url = urljoin(self.base_url, f"/rest/attribute/groups/name/{attribute_group_name}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}")
def attribute_group_add_operation(self, operation_data: Dict[str, Any]) -> SingleAttributeGroupResponse:
"""
Perform an add operation on an attribute group.
Args:
operation_data: Operation details for adding to attribute group
Returns:
SingleAttributeGroupResponse with updated attribute group
"""
return self._make_request(
"POST",
"attribute/groups/operations/add",
SingleAttributeGroupResponse,
json_data=operation_data,
)
# ============= Text Endpoints =============
def get_all_texts(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10,
include_content: bool = False, include_attributes: bool = False
) -> TextListResponse:
"""
Get all texts with optional content and attributes.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of texts per page (default: 10)
include_content: Include text content (default: False)
include_attributes: Include text attributes (default: False)
Returns:
TextListResponse with paginated list of texts
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
if include_content:
params["includeContent"] = "true"
if include_attributes:
params["includeAttributes"] = "true"
return self._make_request(
"GET",
"text",
TextListResponse,
params=params,
)
def create_text(self, text_data: Dict[str, Any]) -> SingleTextResponse:
"""
Create a new text.
Args:
text_data: Text data
Returns:
SingleTextResponse with created text
"""
return self._make_request(
"POST",
"text",
SingleTextResponse,
json_data=text_data,
)
def update_text(self, text_data: Dict[str, Any]) -> SingleTextResponse:
"""
Update a text.
Args:
text_data: Updated text data (must include 'id')
Returns:
SingleTextResponse with updated text
"""
return self._make_request(
"PATCH",
"text",
SingleTextResponse,
json_data=text_data,
)
def create_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkCreateResponse:
"""
Create multiple texts in bulk.
Args:
texts_list: List of text data
Returns:
TextBulkCreateResponse with created texts
"""
return self._make_request(
"POST",
"text/bulk",
TextBulkCreateResponse,
json_data=texts_list,
)
def update_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkUpdateResponse:
"""
Update multiple texts in bulk.
Args:
texts_list: List of text data to update (each must include 'id')
Returns:
TextBulkUpdateResponse with updated texts
"""
return self._make_request(
"PATCH",
"text/bulk",
TextBulkUpdateResponse,
json_data=texts_list,
)
def get_text_by_id(self, text_id: int, lang: Optional[str] = None) -> SingleTextResponse:
"""
Get a text by ID.
Args:
text_id: ID of the text
lang: Language code (optional)
Returns:
SingleTextResponse with text details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"text/{text_id}",
SingleTextResponse,
params=params if params else None,
)
def delete_text(self, text_id: int) -> None:
"""
Delete a text by ID.
Args:
text_id: ID of the text to delete
"""
url = urljoin(self.base_url, f"/rest/text/{text_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete text: {response.status_code}")
```

800
methods_to_add.py
View file

@ -1,800 +0,0 @@
# ============= Product Endpoints =============
def get_all_products(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10, group_id: Optional[int] = None
) -> ProductListResponse:
"""
Get all products with optional group filter.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of products per page (default: 10)
group_id: Optional product group ID to filter products
Returns:
ProductListResponse with paginated list of products
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
if group_id:
params["groupId"] = group_id
return self._make_request(
"GET",
"products",
ProductListResponse,
params=params,
)
def create_product(self, product_data: Dict[str, Any]) -> SingleProductResponse:
"""
Create a new product.
Args:
product_data: Product data
Returns:
SingleProductResponse with created product
"""
return self._make_request(
"POST",
"products",
SingleProductResponse,
json_data=product_data,
)
def update_product(self, product_data: Dict[str, Any]) -> ProductListResponse:
"""
Update a product.
Args:
product_data: Updated product data (must include 'id')
Returns:
ProductListResponse with updated product info
"""
return self._make_request(
"PATCH",
"products",
ProductListResponse,
json_data=product_data,
)
def create_multiple_products(self, products_list: List[Dict[str, Any]]) -> ProductBulkCreateResponse:
"""
Create multiple products in bulk.
Args:
products_list: List of product data
Returns:
ProductBulkCreateResponse with created products
"""
return self._make_request(
"POST",
"products/bulk",
ProductBulkCreateResponse,
json_data=products_list,
)
def update_multiple_products(self, products_list: List[Dict[str, Any]]) -> ProductBulkUpdateResponse:
"""
Update multiple products in bulk.
Args:
products_list: List of product data to update (each must include 'id')
Returns:
ProductBulkUpdateResponse with updated products
"""
return self._make_request(
"PATCH",
"products/bulk",
ProductBulkUpdateResponse,
json_data=products_list,
)
def get_product_by_id(self, product_id: int, lang: Optional[str] = None) -> SingleProductResponse:
"""
Get a product by ID.
Args:
product_id: ID of the product
lang: Language code (optional)
Returns:
SingleProductResponse with product details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"products/{product_id}",
SingleProductResponse,
params=params if params else None,
)
def delete_product(self, product_id: int) -> None:
"""
Delete a product by ID.
Args:
product_id: ID of the product to delete
"""
url = urljoin(self.base_url, f"/rest/products/{product_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete product: {response.status_code}")
def product_operation(self, operation_data: Dict[str, Any]) -> ProductListResponse:
"""
Perform bulk operations on products (copy, move, link, copy-structure).
Args:
operation_data: Operation details including operation type, source, target, etc.
Returns:
ProductListResponse with affected products
"""
return self._make_request(
"POST",
"products/operation",
ProductListResponse,
json_data=operation_data,
)
# ============= Product Group Endpoints =============
def get_all_product_groups(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> ProductGroupListResponse:
"""
Get all product groups.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of groups per page (default: 10)
Returns:
ProductGroupListResponse with paginated list of groups
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"groups",
ProductGroupListResponse,
params=params,
)
def create_product_group(self, group_data: Dict[str, Any]) -> SingleProductGroupResponse:
"""
Create a new product group.
Args:
group_data: Product group data
Returns:
SingleProductGroupResponse with created group
"""
return self._make_request(
"POST",
"groups",
SingleProductGroupResponse,
json_data=group_data,
)
def update_product_group(self, group_data: Dict[str, Any]) -> ProductGroupListResponse:
"""
Update a product group.
Args:
group_data: Updated group data (must include 'id')
Returns:
ProductGroupListResponse with updated group info
"""
return self._make_request(
"PATCH",
"groups",
ProductGroupListResponse,
json_data=group_data,
)
def get_product_group_by_id(self, group_id: int, lang: Optional[str] = None) -> SingleProductGroupResponse:
"""
Get a product group by ID.
Args:
group_id: ID of the product group
lang: Language code (optional)
Returns:
SingleProductGroupResponse with group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"groups/{group_id}",
SingleProductGroupResponse,
params=params if params else None,
)
def delete_product_group(self, group_id: int) -> None:
"""
Delete a product group by ID.
Args:
group_id: ID of the product group to delete
"""
url = urljoin(self.base_url, f"/rest/groups/{group_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete product group: {response.status_code}")
# ============= Tree Group Endpoints =============
def get_all_tree_groups(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> TreeGroupListResponse:
"""
Get all tree groups.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of groups per page (default: 10)
Returns:
TreeGroupListResponse with paginated list of tree groups
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"tree/groups",
TreeGroupListResponse,
params=params,
)
def create_tree_group(self, group_data: Dict[str, Any]) -> SingleTreeGroupResponse:
"""
Create a new tree group.
Args:
group_data: Tree group data
Returns:
SingleTreeGroupResponse with created tree group
"""
return self._make_request(
"POST",
"tree/groups",
SingleTreeGroupResponse,
json_data=group_data,
)
def update_tree_group(self, group_data: Dict[str, Any]) -> SingleTreeGroupResponse:
"""
Update a tree group.
Args:
group_data: Updated tree group data (must include 'id')
Returns:
SingleTreeGroupResponse with updated tree group
"""
return self._make_request(
"PATCH",
"tree/groups",
SingleTreeGroupResponse,
json_data=group_data,
)
def get_tree_group_by_id(self, tree_group_id: int, lang: Optional[str] = None) -> SingleTreeGroupResponse:
"""
Get a tree group by ID.
Args:
tree_group_id: ID of the tree group
lang: Language code (optional)
Returns:
SingleTreeGroupResponse with tree group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"tree/groups/{tree_group_id}",
SingleTreeGroupResponse,
params=params if params else None,
)
def delete_tree_group(self, tree_group_id: int) -> None:
"""
Delete a tree group by ID.
Args:
tree_group_id: ID of the tree group to delete
"""
url = urljoin(self.base_url, f"/rest/tree/groups/{tree_group_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete tree group: {response.status_code}")
# ============= Attribute Endpoints =============
def get_all_attributes(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> AttributeListResponse:
"""
Get all attributes.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of attributes per page (default: 10)
Returns:
AttributeListResponse with paginated list of attributes
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"attributes",
AttributeListResponse,
params=params,
)
def create_attribute(self, attribute_data: Dict[str, Any]) -> SimpleAttributeResponse:
"""
Create a new attribute.
Args:
attribute_data: Attribute data
Returns:
SimpleAttributeResponse with created attribute
"""
return self._make_request(
"POST",
"attributes",
SimpleAttributeResponse,
json_data=attribute_data,
)
def update_attribute(self, attribute_data: Dict[str, Any]) -> SimpleAttributeResponse:
"""
Update an attribute.
Args:
attribute_data: Updated attribute data (must include 'id')
Returns:
SimpleAttributeResponse with updated attribute
"""
return self._make_request(
"PATCH",
"attributes",
SimpleAttributeResponse,
json_data=attribute_data,
)
def create_multiple_attributes(self, attributes_list: List[Dict[str, Any]]) -> AttributeBulkCreateResponse:
"""
Create multiple attributes in bulk.
Args:
attributes_list: List of attribute data
Returns:
AttributeBulkCreateResponse with created attributes
"""
return self._make_request(
"POST",
"attributes/bulk",
AttributeBulkCreateResponse,
json_data=attributes_list,
)
def update_multiple_attributes(self, attributes_list: List[Dict[str, Any]]) -> AttributeBulkUpdateResponse:
"""
Update multiple attributes in bulk.
Args:
attributes_list: List of attribute data to update (each must include 'id')
Returns:
AttributeBulkUpdateResponse with updated attributes
"""
return self._make_request(
"PATCH",
"attributes/bulk",
AttributeBulkUpdateResponse,
json_data=attributes_list,
)
def get_attribute_by_id(self, attribute_id: int, lang: Optional[str] = None) -> SimpleAttributeResponse:
"""
Get an attribute by ID.
Args:
attribute_id: ID of the attribute
lang: Language code (optional)
Returns:
SimpleAttributeResponse with attribute details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attributes/{attribute_id}",
SimpleAttributeResponse,
params=params if params else None,
)
def delete_attribute(self, attribute_id: int) -> None:
"""
Delete an attribute by ID.
Args:
attribute_id: ID of the attribute to delete
"""
url = urljoin(self.base_url, f"/rest/attributes/{attribute_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete attribute: {response.status_code}")
def get_attribute_by_name(self, attribute_name: str, lang: Optional[str] = None) -> AttributeGetByNameResponse:
"""
Get an attribute by name with language-dependent properties.
Args:
attribute_name: Name of the attribute
lang: Language code (optional)
Returns:
AttributeGetByNameResponse with attribute details and languageDependents
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attributes/name/{attribute_name}",
AttributeGetByNameResponse,
params=params if params else None,
)
# ============= Attribute Group Endpoints =============
def get_all_attribute_groups(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
) -> AttributeGroupListResponse:
"""
Get all attribute groups.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of groups per page (default: 10)
Returns:
AttributeGroupListResponse with paginated list of attribute groups
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
"attribute/groups",
AttributeGroupListResponse,
params=params,
)
def create_attribute_group(self, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse:
"""
Create a new attribute group.
Args:
group_data: Attribute group data
Returns:
SingleAttributeGroupResponse with created attribute group
"""
return self._make_request(
"POST",
"attribute/groups",
SingleAttributeGroupResponse,
json_data=group_data,
)
def get_attribute_group_by_id(self, attribute_group_id: int, lang: Optional[str] = None) -> SingleAttributeGroupResponse:
"""
Get an attribute group by ID.
Args:
attribute_group_id: ID of the attribute group
lang: Language code (optional)
Returns:
SingleAttributeGroupResponse with attribute group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attribute/groups/{attribute_group_id}",
SingleAttributeGroupResponse,
params=params if params else None,
)
def delete_attribute_group(self, attribute_group_id: int) -> None:
"""
Delete an attribute group by ID.
Args:
attribute_group_id: ID of the attribute group to delete
"""
url = urljoin(self.base_url, f"/rest/attribute/groups/{attribute_group_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}")
def get_attribute_group_by_name(self, attribute_group_name: str, lang: Optional[str] = None) -> SingleAttributeGroupResponse:
"""
Get an attribute group by name.
Args:
attribute_group_name: Name of the attribute group
lang: Language code (optional)
Returns:
SingleAttributeGroupResponse with attribute group details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"attribute/groups/name/{attribute_group_name}",
SingleAttributeGroupResponse,
params=params if params else None,
)
def update_attribute_group_by_name(self, attribute_group_name: str, group_data: Dict[str, Any]) -> SingleAttributeGroupResponse:
"""
Update an attribute group by name.
Args:
attribute_group_name: Name of the attribute group to update
group_data: Updated attribute group data
Returns:
SingleAttributeGroupResponse with updated attribute group
"""
return self._make_request(
"PATCH",
f"attribute/groups/name/{attribute_group_name}",
SingleAttributeGroupResponse,
json_data=group_data,
)
def delete_attribute_group_by_name(self, attribute_group_name: str) -> None:
"""
Delete an attribute group by name.
Args:
attribute_group_name: Name of the attribute group to delete
"""
url = urljoin(self.base_url, f"/rest/attribute/groups/name/{attribute_group_name}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete attribute group: {response.status_code}")
def attribute_group_add_operation(self, operation_data: Dict[str, Any]) -> SingleAttributeGroupResponse:
"""
Perform an add operation on an attribute group.
Args:
operation_data: Operation details for adding to attribute group
Returns:
SingleAttributeGroupResponse with updated attribute group
"""
return self._make_request(
"POST",
"attribute/groups/operations/add",
SingleAttributeGroupResponse,
json_data=operation_data,
)
# ============= Text Endpoints =============
def get_all_texts(
self, lang: Optional[str] = None, page: int = 1, limit: int = 10,
include_content: bool = False, include_attributes: bool = False
) -> TextListResponse:
"""
Get all texts with optional content and attributes.
Args:
lang: Language code (optional)
page: Page number (default: 1)
limit: Number of texts per page (default: 10)
include_content: Include text content (default: False)
include_attributes: Include text attributes (default: False)
Returns:
TextListResponse with paginated list of texts
"""
params: Dict[str, Any] = {"page": page, "limit": limit}
if lang:
params["lang"] = lang
if include_content:
params["includeContent"] = "true"
if include_attributes:
params["includeAttributes"] = "true"
return self._make_request(
"GET",
"text",
TextListResponse,
params=params,
)
def create_text(self, text_data: Dict[str, Any]) -> SingleTextResponse:
"""
Create a new text.
Args:
text_data: Text data
Returns:
SingleTextResponse with created text
"""
return self._make_request(
"POST",
"text",
SingleTextResponse,
json_data=text_data,
)
def update_text(self, text_data: Dict[str, Any]) -> SingleTextResponse:
"""
Update a text.
Args:
text_data: Updated text data (must include 'id')
Returns:
SingleTextResponse with updated text
"""
return self._make_request(
"PATCH",
"text",
SingleTextResponse,
json_data=text_data,
)
def create_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkCreateResponse:
"""
Create multiple texts in bulk.
Args:
texts_list: List of text data
Returns:
TextBulkCreateResponse with created texts
"""
return self._make_request(
"POST",
"text/bulk",
TextBulkCreateResponse,
json_data=texts_list,
)
def update_multiple_texts(self, texts_list: List[Dict[str, Any]]) -> TextBulkUpdateResponse:
"""
Update multiple texts in bulk.
Args:
texts_list: List of text data to update (each must include 'id')
Returns:
TextBulkUpdateResponse with updated texts
"""
return self._make_request(
"PATCH",
"text/bulk",
TextBulkUpdateResponse,
json_data=texts_list,
)
def get_text_by_id(self, text_id: int, lang: Optional[str] = None) -> SingleTextResponse:
"""
Get a text by ID.
Args:
text_id: ID of the text
lang: Language code (optional)
Returns:
SingleTextResponse with text details
"""
params: Dict[str, Any] = {}
if lang:
params["lang"] = lang
return self._make_request(
"GET",
f"text/{text_id}",
SingleTextResponse,
params=params if params else None,
)
def delete_text(self, text_id: int) -> None:
"""
Delete a text by ID.
Args:
text_id: ID of the text to delete
"""
url = urljoin(self.base_url, f"/rest/text/{text_id}")
params = None
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
params = self.auth.get_url_parameters()
response = self.session.delete(url, params=params, timeout=self.timeout)
if response.status_code >= 400:
raise ElytraAPIError(f"Failed to delete text: {response.status_code}")