Refactor models in the Lobster PIM Legacy REST API

- Organized model imports in __init__.py for better structure and readability.
- Grouped related models into separate files for attributes, attribute groups, products, protocols, and tree groups.
- Removed duplicate imports and ensured all necessary models are included.
- Enhanced code maintainability by separating concerns and improving the overall organization of the models package.
This commit is contained in:
claudi 2026-03-24 15:35:46 +01:00
parent 68f0b76feb
commit 310a4fe2f8
2 changed files with 1053 additions and 55 deletions

View file

@ -0,0 +1,999 @@
"""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"
)
# ============================================================================
# PRODUCT MODELS
# ============================================================================
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
# ============================================================================
# PRODUCT GROUP MODELS
# ============================================================================
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
# ============================================================================
# TREE GROUP MODELS
# ============================================================================
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
# ============================================================================
# ATTRIBUTE MODELS
# ============================================================================
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")
# ============================================================================
# ATTRIBUTE GROUP MODELS
# ============================================================================
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
# ============================================================================
# TEXT MODELS
# ============================================================================
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")
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()

View file

@ -1,7 +1,28 @@
"""Models package for Lobster PIM Legacy REST API"""
# Shared models
from .shared import AttributeResponse, ErrorResponse, PaginationLinks
# Attribute group models
from .attribute_groups import (
AttributeGroupBulkCreateResponse,
AttributeGroupHierarchyNode,
AttributeGroupHierarchyResponse,
AttributeGroupListResponse,
AttributeGroupValidFor,
SingleAttributeGroupResponse,
SingleNewAttributeGroupRequestBody,
SingleUpdateAttributeGroupRequestBody,
)
# Attribute models
from .attributes import (
AttributeBulkCreateResponse,
AttributeBulkUpdateResponse,
AttributeGetByNameResponse,
AttributeListResponse,
SimpleAttributeResponse,
SingleNewAttributeRequestBody,
SingleUpdateAttributeRequestBody,
)
# Hierarchy models
from .hierarchy import HierarchyNode
@ -16,15 +37,6 @@ from .jobs import (
JobOverviewResponse,
)
# Protocol models
from .protocols import (
ProtocolCategoryInfo,
ProtocolCategoryListResponse,
ProtocolEntry,
ProtocolInfo,
ProtocolListResponse,
)
# Media models
from .media import (
MediaBulkCreateResponse,
@ -36,6 +48,18 @@ from .media import (
SingleUpdateMediaRequestBody,
)
# Product group models
from .product_groups import (
ProductGroupBulkCreateResponse,
ProductGroupBulkUpdateResponse,
ProductGroupHierarchyNode,
ProductGroupHierarchyResponse,
ProductGroupListResponse,
SingleNewProductGroupRequestBody,
SingleProductGroupResponse,
SingleUpdateProductGroupRequestBody,
)
# Product models
from .products import (
ProductAttributeResponse,
@ -50,16 +74,26 @@ from .products import (
SingleUpdateProductRequestBody,
)
# Product group models
from .product_groups import (
ProductGroupBulkCreateResponse,
ProductGroupBulkUpdateResponse,
ProductGroupHierarchyNode,
ProductGroupHierarchyResponse,
ProductGroupListResponse,
SingleNewProductGroupRequestBody,
SingleProductGroupResponse,
SingleUpdateProductGroupRequestBody,
# Protocol models
from .protocols import (
ProtocolCategoryInfo,
ProtocolCategoryListResponse,
ProtocolEntry,
ProtocolInfo,
ProtocolListResponse,
)
from .shared import AttributeResponse, ErrorResponse, PaginationLinks
# Text models
from .text import (
SingleNewTextRequestBody,
SingleTextResponse,
SingleUpdateTextRequestBody,
TextBulkCreateResponse,
TextBulkUpdateResponse,
TextContentRequestBody,
TextContentResponse,
TextListResponse,
)
# Tree group models
@ -74,41 +108,6 @@ from .tree_groups import (
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",