feat: Add media and hierarchy endpoints with corresponding models to support media management
This commit is contained in:
parent
dce4ac5643
commit
ef53746129
2 changed files with 567 additions and 8 deletions
|
|
@ -14,19 +14,31 @@ from ..exceptions import (
|
|||
)
|
||||
from .auth import AuthMethod, RestApiAuth
|
||||
from .models import (
|
||||
AttributeGroupHierarchyResponse,
|
||||
AttributeResponse,
|
||||
JobControlRequest,
|
||||
JobControlResponse,
|
||||
JobDetailInfo,
|
||||
JobExecutionResponse,
|
||||
JobInfo,
|
||||
JobOverviewResponse,
|
||||
MediaBulkCreateResponse,
|
||||
MediaBulkUpdateResponse,
|
||||
MediaFileResponse,
|
||||
MediaListResponse,
|
||||
ProductGroupHierarchyResponse,
|
||||
ProductHierarchyResponse,
|
||||
ProtocolCategoryInfo,
|
||||
ProtocolCategoryListResponse,
|
||||
ProtocolInfo,
|
||||
ProtocolListResponse,
|
||||
SingleMediaResponse,
|
||||
SingleNewMediaRequestBody,
|
||||
SingleUpdateMediaRequestBody,
|
||||
TreeGroupHierarchyResponse,
|
||||
)
|
||||
|
||||
T = TypeVar('T', bound=BaseModel)
|
||||
T = TypeVar("T", bound=BaseModel)
|
||||
|
||||
|
||||
class LobsterRestApiClient:
|
||||
|
|
@ -425,6 +437,314 @@ class LobsterRestApiClient:
|
|||
params={"limit": limit},
|
||||
)
|
||||
|
||||
# ============= Media Endpoints =============
|
||||
|
||||
def get_all_media(
|
||||
self, lang: Optional[str] = None, page: int = 1, limit: int = 10
|
||||
) -> MediaListResponse:
|
||||
"""
|
||||
Get all media descriptors with pagination.
|
||||
|
||||
Args:
|
||||
lang: Language code (optional)
|
||||
page: Page number (default: 1)
|
||||
limit: Number of media items per page (default: 10)
|
||||
|
||||
Returns:
|
||||
MediaListResponse containing paginated list of media items
|
||||
"""
|
||||
params: Dict[str, Any] = {"page": page, "limit": limit}
|
||||
if lang:
|
||||
params["lang"] = lang
|
||||
|
||||
return self._make_request(
|
||||
"GET",
|
||||
"media",
|
||||
MediaListResponse,
|
||||
params=params,
|
||||
)
|
||||
|
||||
def get_media_by_id(self, media_id: int, lang: Optional[str] = None) -> SingleMediaResponse:
|
||||
"""
|
||||
Get a media descriptor by ID.
|
||||
|
||||
Args:
|
||||
media_id: ID of the media
|
||||
lang: Language code (optional)
|
||||
|
||||
Returns:
|
||||
SingleMediaResponse with media details
|
||||
"""
|
||||
params: Dict[str, Any] = {}
|
||||
if lang:
|
||||
params["lang"] = lang
|
||||
|
||||
return self._make_request(
|
||||
"GET",
|
||||
f"media/{media_id}",
|
||||
SingleMediaResponse,
|
||||
params=params if params else None,
|
||||
)
|
||||
|
||||
def create_media(self, media_data: Dict[str, Any]) -> SingleMediaResponse:
|
||||
"""
|
||||
Create a new media descriptor.
|
||||
|
||||
Args:
|
||||
media_data: Media descriptor data
|
||||
|
||||
Returns:
|
||||
SingleMediaResponse with created media
|
||||
"""
|
||||
return self._make_request(
|
||||
"POST",
|
||||
"media",
|
||||
SingleMediaResponse,
|
||||
json_data=media_data,
|
||||
)
|
||||
|
||||
def update_media(self, media_data: Dict[str, Any]) -> SingleMediaResponse:
|
||||
"""
|
||||
Update a media descriptor.
|
||||
|
||||
Args:
|
||||
media_data: Updated media descriptor data (must include 'id')
|
||||
|
||||
Returns:
|
||||
SingleMediaResponse with updated media
|
||||
"""
|
||||
return self._make_request(
|
||||
"PATCH",
|
||||
"media",
|
||||
SingleMediaResponse,
|
||||
json_data=media_data,
|
||||
)
|
||||
|
||||
def create_multiple_media(self, media_list: List[Dict[str, Any]]) -> MediaBulkCreateResponse:
|
||||
"""
|
||||
Create multiple media descriptors in bulk.
|
||||
|
||||
Args:
|
||||
media_list: List of media descriptor data
|
||||
|
||||
Returns:
|
||||
MediaBulkCreateResponse with created media items
|
||||
"""
|
||||
return self._make_request(
|
||||
"POST",
|
||||
"media/bulk",
|
||||
MediaBulkCreateResponse,
|
||||
json_data=media_list,
|
||||
)
|
||||
|
||||
def update_multiple_media(self, media_list: List[Dict[str, Any]]) -> MediaBulkUpdateResponse:
|
||||
"""
|
||||
Update multiple media descriptors in bulk.
|
||||
|
||||
Args:
|
||||
media_list: List of media descriptor data to update (each must include 'id')
|
||||
|
||||
Returns:
|
||||
MediaBulkUpdateResponse with updated media items
|
||||
"""
|
||||
return self._make_request(
|
||||
"PATCH",
|
||||
"media/bulk",
|
||||
MediaBulkUpdateResponse,
|
||||
json_data=media_list,
|
||||
)
|
||||
|
||||
def delete_media(self, media_id: int) -> None:
|
||||
"""
|
||||
Delete a media descriptor by ID.
|
||||
|
||||
Args:
|
||||
media_id: ID of the media to delete
|
||||
"""
|
||||
url = urljoin(self.base_url, f"/rest/media/{media_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 media: {response.status_code}")
|
||||
|
||||
def upload_media_file(
|
||||
self,
|
||||
file_path: str,
|
||||
media_id: int,
|
||||
mam_system: str,
|
||||
language_code: str = "independent",
|
||||
) -> MediaFileResponse:
|
||||
"""
|
||||
Upload a media file for a media descriptor.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file to upload
|
||||
media_id: ID of the media descriptor
|
||||
mam_system: MAM system code (fs, sixomc, cumulus, etc.)
|
||||
language_code: Language code for the file (default: independent)
|
||||
|
||||
Returns:
|
||||
MediaFileResponse with uploaded file metadata
|
||||
"""
|
||||
url = urljoin(self.base_url, "/rest/media/file")
|
||||
|
||||
with open(file_path, "rb") as file:
|
||||
files = {"file": file}
|
||||
data = {
|
||||
"mediaId": media_id,
|
||||
"mamSystem": mam_system,
|
||||
"languageCode": language_code,
|
||||
}
|
||||
# Add authentication credentials
|
||||
data.update(self.auth.get_json_body_params())
|
||||
|
||||
response = self.session.post(
|
||||
url,
|
||||
files=files,
|
||||
data=data,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
return self._handle_response(response, MediaFileResponse)
|
||||
|
||||
def get_media_content(self, media_file_id: int) -> bytes:
|
||||
"""
|
||||
Download the binary content of a media file.
|
||||
|
||||
Args:
|
||||
media_file_id: ID of the media file
|
||||
|
||||
Returns:
|
||||
Binary content of the media file
|
||||
"""
|
||||
url = urljoin(self.base_url, f"/rest/media/file/{media_file_id}")
|
||||
params = None
|
||||
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
|
||||
params = self.auth.get_url_parameters()
|
||||
|
||||
response = self.session.get(url, params=params, timeout=self.timeout)
|
||||
if response.status_code >= 400:
|
||||
raise ElytraAPIError(f"Failed to download media: {response.status_code}")
|
||||
|
||||
return response.content
|
||||
|
||||
def delete_media_file(self, media_file_id: int) -> None:
|
||||
"""
|
||||
Delete a media file by ID.
|
||||
|
||||
Args:
|
||||
media_file_id: ID of the media file to delete
|
||||
"""
|
||||
url = urljoin(self.base_url, f"/rest/media/file/{media_file_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 media file: {response.status_code}")
|
||||
|
||||
# ============= Hierarchy Endpoints =============
|
||||
|
||||
def get_product_hierarchy(self, product_id: int, depth: int = 10) -> ProductHierarchyResponse:
|
||||
"""
|
||||
Get the hierarchy of a product.
|
||||
|
||||
Args:
|
||||
product_id: ID of the product
|
||||
depth: Depth of the hierarchy (default: 10)
|
||||
|
||||
Returns:
|
||||
ProductHierarchyResponse with hierarchical product structure
|
||||
"""
|
||||
params = {"depth": depth}
|
||||
return self._make_request(
|
||||
"GET",
|
||||
f"products/{product_id}/hierarchy",
|
||||
ProductHierarchyResponse,
|
||||
params=params,
|
||||
)
|
||||
|
||||
def get_product_group_hierarchy(
|
||||
self, group_id: int, depth: int = 10
|
||||
) -> ProductGroupHierarchyResponse:
|
||||
"""
|
||||
Get the hierarchy of a product group.
|
||||
|
||||
Args:
|
||||
group_id: ID of the product group
|
||||
depth: Depth of the hierarchy (default: 10)
|
||||
|
||||
Returns:
|
||||
ProductGroupHierarchyResponse with hierarchical group structure
|
||||
"""
|
||||
params = {"depth": depth}
|
||||
return self._make_request(
|
||||
"GET",
|
||||
f"groups/{group_id}/hierarchy",
|
||||
ProductGroupHierarchyResponse,
|
||||
params=params,
|
||||
)
|
||||
|
||||
def get_tree_group_hierarchy(self, depth: int = 10) -> List[TreeGroupHierarchyResponse]:
|
||||
"""
|
||||
Get the hierarchy of tree groups.
|
||||
|
||||
Args:
|
||||
depth: Depth of the hierarchy (default: 10)
|
||||
|
||||
Returns:
|
||||
List of TreeGroupHierarchyResponse with hierarchical tree structure
|
||||
"""
|
||||
params = {"depth": depth}
|
||||
url = urljoin(self.base_url, "/rest/tree/groups/hierarchy")
|
||||
|
||||
if self.auth.auth_method == AuthMethod.USERNAME_PASSWORD:
|
||||
params.update(self.auth.get_url_parameters())
|
||||
|
||||
response = self.session.request(
|
||||
method="GET",
|
||||
url=url,
|
||||
params=params,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
if response.status_code >= 400:
|
||||
raise ElytraAPIError(f"Failed to get tree hierarchy: {response.status_code}")
|
||||
|
||||
try:
|
||||
data = response.json()
|
||||
if isinstance(data, list):
|
||||
return [TreeGroupHierarchyResponse.model_validate(item) for item in data]
|
||||
else:
|
||||
return [TreeGroupHierarchyResponse.model_validate(data)]
|
||||
except Exception as e:
|
||||
raise ElytraValidationError(f"Failed to parse tree hierarchy: {str(e)}")
|
||||
|
||||
def get_attribute_group_hierarchy(
|
||||
self, attribute_group_id: int, depth: int = 10
|
||||
) -> AttributeGroupHierarchyResponse:
|
||||
"""
|
||||
Get the hierarchy of an attribute group.
|
||||
|
||||
Args:
|
||||
attribute_group_id: ID of the attribute group
|
||||
depth: Depth of the hierarchy (default: 10)
|
||||
|
||||
Returns:
|
||||
AttributeGroupHierarchyResponse with hierarchical attribute group structure
|
||||
"""
|
||||
params = {"depth": depth}
|
||||
return self._make_request(
|
||||
"GET",
|
||||
f"attribute/groups/hierarchy/{attribute_group_id}",
|
||||
AttributeGroupHierarchyResponse,
|
||||
params=params,
|
||||
)
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the session and clean up resources."""
|
||||
self.session.close()
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ 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")
|
||||
|
|
@ -22,12 +23,14 @@ class JobInfo(BaseModel):
|
|||
|
||||
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")
|
||||
|
|
@ -35,6 +38,7 @@ class JobOverviewResponse(BaseModel):
|
|||
|
||||
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")
|
||||
|
|
@ -44,12 +48,15 @@ class JobExecutionResponse(BaseModel):
|
|||
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)")
|
||||
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")
|
||||
|
|
@ -61,9 +68,7 @@ class JobControlRequest(BaseModel):
|
|||
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"
|
||||
)
|
||||
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)"
|
||||
)
|
||||
|
|
@ -71,6 +76,7 @@ class JobControlRequest(BaseModel):
|
|||
|
||||
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")
|
||||
|
|
@ -80,6 +86,7 @@ class JobControlResponse(BaseModel):
|
|||
|
||||
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")
|
||||
|
|
@ -87,6 +94,7 @@ class ProtocolEntry(BaseModel):
|
|||
|
||||
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")
|
||||
|
|
@ -97,13 +105,12 @@ class ProtocolInfo(BaseModel):
|
|||
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"
|
||||
)
|
||||
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")
|
||||
|
|
@ -111,6 +118,7 @@ class ProtocolListResponse(BaseModel):
|
|||
|
||||
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")
|
||||
|
|
@ -118,12 +126,243 @@ class ProtocolCategoryInfo(BaseModel):
|
|||
|
||||
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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue