c73e39f6e4
- Fix Python formatting (sync no longer strips blank lines from .py files) - Remove test_community_build.py (tests private sync module) - Fix ruff warnings in test files - Add CODE_OF_CONDUCT.md - Add .github/PULL_REQUEST_TEMPLATE.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
260 lines
8.1 KiB
Python
260 lines
8.1 KiB
Python
"""
|
|
Bulk Operations Schemas
|
|
|
|
Pydantic models for WordPress bulk operations including:
|
|
- Bulk updates for posts/products
|
|
- Bulk deletions
|
|
- Bulk category/tag assignments
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
class BulkUpdatePostsParams(BaseModel):
|
|
"""Parameters for bulk updating posts"""
|
|
|
|
post_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of post IDs to update (max 100)"
|
|
)
|
|
updates: dict[str, Any] = Field(
|
|
..., description="Fields to update (status, author_id, categories, tags, etc.)"
|
|
)
|
|
|
|
@classmethod
|
|
@field_validator("post_ids")
|
|
def validate_post_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All post IDs must be positive integers")
|
|
return v
|
|
|
|
@classmethod
|
|
@field_validator("updates")
|
|
def validate_updates(cls, v):
|
|
"""Validate update fields are allowed"""
|
|
allowed_fields = {
|
|
"status",
|
|
"title",
|
|
"content",
|
|
"excerpt",
|
|
"author",
|
|
"categories",
|
|
"tags",
|
|
"featured_media",
|
|
"comment_status",
|
|
"ping_status",
|
|
"sticky",
|
|
"format",
|
|
"meta",
|
|
}
|
|
|
|
invalid_fields = set(v.keys()) - allowed_fields
|
|
if invalid_fields:
|
|
raise ValueError(f"Invalid update fields: {', '.join(invalid_fields)}")
|
|
|
|
return v
|
|
|
|
|
|
class BulkDeletePostsParams(BaseModel):
|
|
"""Parameters for bulk deleting posts"""
|
|
|
|
post_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of post IDs to delete (max 100)"
|
|
)
|
|
force: bool = Field(default=False, description="Force permanent deletion (bypass trash)")
|
|
|
|
@classmethod
|
|
@field_validator("post_ids")
|
|
def validate_post_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All post IDs must be positive integers")
|
|
return v
|
|
|
|
|
|
class BulkUpdateProductsParams(BaseModel):
|
|
"""Parameters for bulk updating WooCommerce products"""
|
|
|
|
product_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of product IDs to update (max 100)"
|
|
)
|
|
updates: dict[str, Any] = Field(
|
|
..., description="Fields to update (price, stock_quantity, status, etc.)"
|
|
)
|
|
|
|
@classmethod
|
|
@field_validator("product_ids")
|
|
def validate_product_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All product IDs must be positive integers")
|
|
return v
|
|
|
|
@classmethod
|
|
@field_validator("updates")
|
|
def validate_updates(cls, v):
|
|
"""Validate update fields are allowed"""
|
|
allowed_fields = {
|
|
"name",
|
|
"status",
|
|
"featured",
|
|
"catalog_visibility",
|
|
"description",
|
|
"short_description",
|
|
"sku",
|
|
"price",
|
|
"regular_price",
|
|
"sale_price",
|
|
"stock_quantity",
|
|
"stock_status",
|
|
"manage_stock",
|
|
"categories",
|
|
"tags",
|
|
"images",
|
|
"attributes",
|
|
"meta_data",
|
|
}
|
|
|
|
invalid_fields = set(v.keys()) - allowed_fields
|
|
if invalid_fields:
|
|
raise ValueError(f"Invalid update fields: {', '.join(invalid_fields)}")
|
|
|
|
return v
|
|
|
|
|
|
class BulkDeleteProductsParams(BaseModel):
|
|
"""Parameters for bulk deleting products"""
|
|
|
|
product_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of product IDs to delete (max 100)"
|
|
)
|
|
force: bool = Field(default=False, description="Force permanent deletion")
|
|
|
|
@classmethod
|
|
@field_validator("product_ids")
|
|
def validate_product_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All product IDs must be positive integers")
|
|
return v
|
|
|
|
|
|
class BulkAssignCategoriesParams(BaseModel):
|
|
"""Parameters for bulk assigning categories"""
|
|
|
|
item_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of post/product IDs (max 100)"
|
|
)
|
|
category_ids: list[int] = Field(..., min_length=1, description="List of category IDs to assign")
|
|
replace: bool = Field(
|
|
default=False, description="Replace existing categories (true) or append (false)"
|
|
)
|
|
item_type: str = Field(default="post", description="Type of items: 'post' or 'product'")
|
|
|
|
@classmethod
|
|
@field_validator("item_ids", "category_ids")
|
|
def validate_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All IDs must be positive integers")
|
|
return v
|
|
|
|
@classmethod
|
|
@field_validator("item_type")
|
|
def validate_item_type(cls, v):
|
|
"""Validate item type"""
|
|
if v not in ["post", "product"]:
|
|
raise ValueError("item_type must be 'post' or 'product'")
|
|
return v
|
|
|
|
|
|
class BulkAssignTagsParams(BaseModel):
|
|
"""Parameters for bulk assigning tags"""
|
|
|
|
item_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of post/product IDs (max 100)"
|
|
)
|
|
tag_ids: list[int] = Field(..., min_length=1, description="List of tag IDs to assign")
|
|
replace: bool = Field(
|
|
default=False, description="Replace existing tags (true) or append (false)"
|
|
)
|
|
item_type: str = Field(default="post", description="Type of items: 'post' or 'product'")
|
|
|
|
@classmethod
|
|
@field_validator("item_ids", "tag_ids")
|
|
def validate_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All IDs must be positive integers")
|
|
return v
|
|
|
|
@classmethod
|
|
@field_validator("item_type")
|
|
def validate_item_type(cls, v):
|
|
"""Validate item type"""
|
|
if v not in ["post", "product"]:
|
|
raise ValueError("item_type must be 'post' or 'product'")
|
|
return v
|
|
|
|
|
|
class BulkUpdateMediaParams(BaseModel):
|
|
"""Parameters for bulk updating media items"""
|
|
|
|
media_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of media IDs to update (max 100)"
|
|
)
|
|
updates: dict[str, Any] = Field(
|
|
..., description="Fields to update (alt_text, title, caption, description)"
|
|
)
|
|
|
|
@classmethod
|
|
@field_validator("media_ids")
|
|
def validate_media_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All media IDs must be positive integers")
|
|
return v
|
|
|
|
@classmethod
|
|
@field_validator("updates")
|
|
def validate_updates(cls, v):
|
|
"""Validate update fields are allowed"""
|
|
allowed_fields = {"title", "alt_text", "caption", "description", "meta"}
|
|
|
|
invalid_fields = set(v.keys()) - allowed_fields
|
|
if invalid_fields:
|
|
raise ValueError(f"Invalid update fields: {', '.join(invalid_fields)}")
|
|
|
|
return v
|
|
|
|
|
|
class BulkDeleteMediaParams(BaseModel):
|
|
"""Parameters for bulk deleting media items"""
|
|
|
|
media_ids: list[int] = Field(
|
|
..., min_length=1, max_length=100, description="List of media IDs to delete (max 100)"
|
|
)
|
|
force: bool = Field(
|
|
default=True, description="Force permanent deletion (media can't be trashed)"
|
|
)
|
|
|
|
@classmethod
|
|
@field_validator("media_ids")
|
|
def validate_media_ids(cls, v):
|
|
"""Ensure all IDs are positive"""
|
|
if any(id <= 0 for id in v):
|
|
raise ValueError("All media IDs must be positive integers")
|
|
return v
|
|
|
|
|
|
class BulkOperationResult(BaseModel):
|
|
"""Result of a bulk operation"""
|
|
|
|
success_count: int = Field(description="Number of successful operations")
|
|
failed_count: int = Field(description="Number of failed operations")
|
|
total: int = Field(description="Total items processed")
|
|
failed_ids: list[int] = Field(default=[], description="IDs that failed to process")
|
|
errors: list[dict[str, Any]] = Field(default=[], description="Detailed error information")
|