Files
office_translator/services/providers/schemas.py
2026-03-07 11:42:58 +01:00

121 lines
4.1 KiB
Python

"""
Pydantic models for translation provider request/response schemas.
"""
import re
from typing import Optional, List
from pydantic import BaseModel, ConfigDict, Field, field_validator
LANGUAGE_CODE_PATTERN = re.compile(r"^[a-z]{2}(-[A-Z]{2})?$|^auto$")
class TranslationRequest(BaseModel):
"""Request model for translation operations."""
text: str = Field(..., description="Text to translate")
target_language: str = Field(
..., description="Target language code (e.g., 'en', 'fr', 'es')"
)
source_language: str = Field(
default="auto", description="Source language code (default: auto-detect)"
)
metadata: Optional[dict] = Field(
default=None,
description="Optional metadata for provider-specific options (e.g., custom_prompt)",
)
@field_validator("target_language", "source_language")
@classmethod
def validate_language_code(cls, v: str) -> str:
if not LANGUAGE_CODE_PATTERN.match(v):
raise ValueError(
f"Invalid language code '{v}'. Expected format: 'xx' or 'xx-XX' (e.g., 'en', 'fr', 'en-US')"
)
return v
class TranslationResponse(BaseModel):
"""Response model for translation operations."""
translated_text: str = Field(..., description="Translated text")
provider_name: str = Field(
..., description="Name of the provider that performed the translation"
)
from_cache: bool = Field(
default=False, description="Whether the result came from cache"
)
source_language: Optional[str] = Field(
default=None, description="Detected or specified source language"
)
error: Optional[str] = Field(
default=None, description="Error message if translation failed"
)
error_code: Optional[str] = Field(
default=None, description="Error code for programmatic error handling"
)
error_details: Optional[dict] = Field(
default=None, description="Additional error details"
)
@property
def success(self) -> bool:
"""Check if translation was successful."""
return self.error is None
def to_error_dict(self) -> dict:
"""Convert error to dictionary format for API responses."""
if self.error is None:
return {}
result = {
"error": self.error_code or "UNKNOWN_ERROR",
"message": self.error,
}
if self.error_details:
result["details"] = self.error_details
return result
class BatchTranslationRequest(BaseModel):
"""Request model for batch translation operations."""
texts: List[str] = Field(..., description="List of texts to translate")
target_language: str = Field(..., description="Target language code")
source_language: str = Field(
default="auto", description="Source language code (default: auto-detect)"
)
class BatchTranslationResponse(BaseModel):
"""Response model for batch translation operations."""
translated_texts: List[str] = Field(..., description="List of translated texts")
provider_name: str = Field(
..., description="Name of the provider that performed the translations"
)
from_cache_count: int = Field(default=0, description="Number of results from cache")
class ProviderHealthStatus(BaseModel):
"""Health status model for a translation provider."""
model_config = ConfigDict(protected_namespaces=())
name: str = Field(..., description="Provider name")
available: bool = Field(..., description="Whether the provider is available")
latency_ms: Optional[float] = Field(
default=None, description="Response latency in milliseconds"
)
error: Optional[str] = Field(
default=None, description="Error message if unavailable"
)
last_check: Optional[str] = Field(
default=None, description="ISO timestamp of last health check"
)
model: Optional[str] = Field(
default=None, description="Model name (e.g. for LLM providers)"
)
model_available: Optional[bool] = Field(
default=None, description="Whether the configured model is available"
)