pythonProject2/schemas.py - lab4 - Pepegator/task-api

pythonProject2/schemas.py
91 строка3 KB
Pol136
add rabbitmq
30 дек 2025, 17:44
30 дек 2025, 17:447c8d717
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
import uuid
from typing import Dict, Any, Optional

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)

class UserLogin(BaseModel):
    username: str
    password: str

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime
    
    class Config:
        from_attributes = True

class TokenResponse(BaseModel):
    access_token: str
    token_type: str
    expires_in: int

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    priority: str = Field(default="medium", pattern="^(low|medium|high)$")

class TaskUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    priority: Optional[str] = None
    completed_at: Optional[datetime] = None

class TaskResponse(BaseModel):
    id: int
    owner_id: int
    title: str
    description: Optional[str]
    status: str
    priority: str
    created_at: datetime
    updated_at: datetime
    completed_at: Optional[datetime]
    
    class Config:
        from_attributes = True


class AuthModel(BaseModel):
    api_key: str = Field(..., description="API ключ клиента")
    signature: Optional[str] = Field(None)

class RabbitMQRequestMessage(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    version: str = Field(default="v1")
    action: str = Field(...)
    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
    data: Dict[str, Any] = Field(default_factory=dict)
    auth: AuthModel = Field(...)
    reply_to: Optional[str] = Field(None)
    timeout_ms: int = Field(default=5000)

class ErrorDetail(BaseModel):
    type: str = Field(...)
    message: str = Field(...)
    details: Optional[Dict[str, Any]] = Field(None)

class RabbitMQResponseMessage(BaseModel):
    correlation_id: str = Field(...)
    status: str = Field(...)
    code: int = Field(...)
    data: Optional[Dict[str, Any]] = Field(None)
    error: Optional[ErrorDetail] = Field(None)
    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
    processing_time_ms: Optional[int] = Field(None)

class DLQMessage(BaseModel):
    request_id: str
    original_message: Dict[str, Any]
    error_type: str
    error_message: str
    retry_count: int
    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())