FullstackAPI/API v4(test)/main-1.py - dev - Xton/Case_13_HTML_Newsletter_Builder

FullstackAPI/API v4(test)/main-1.py
157 строк5 KB
Xton
create: main-1.py, delete: main.py
12 мар 2026, 18:54
12 мар 2026, 18:54cb80ee5
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
import random
from enum import Enum

app = FastAPI(
    title="Email API",
    description=" API для email-рассылок",
    version="1.0.0"
)

# Контракт данных
class CampaignStatus(str, Enum):
    queued = "queued"
    sending = "sending" 
    completed = "completed"
    failed = "failed"
    not_found = "not_found"

class PriorityLevel(str, Enum):
    low = "low"
    normal = "normal"
    high = "high"
#ПЕРВЫЕ JSON КОНТРАКТЫ
# КОНТРАКТЫ ДАННЫХ - POST /mail/send
class MailSendRequest(BaseModel):
    campaign_id: str = Field(..., min_length=3, max_length=50, description="Уникальный ID кампании")
    client_id: str = Field(..., min_length=3, max_length=50, description="ID клиента")
    recipients: List[str] = Field(..., min_items=1, max_items=1000, description="Список email")
    subject: str = Field(..., max_length=200, description="Тема письма")
    priority: PriorityLevel = Field(default="normal", description="Приоритет")

class MailSendResponse(BaseModel):
    campaign_id: str
    client_id: str
    status: CampaignStatus
    error: Optional[str] = None
    queued_at: str  # ISO datetime

# КОНТРАКТЫ ДАННЫХ - GET /campaigns/{id}/status  
class CampaignStatusResponse(BaseModel):
    campaign_id: str
    client_id: Optional[str] = None
    status: CampaignStatus
    error: Optional[str] = None
    progress: Optional[int] = Field(None, ge=0, le=100)  # 0-100%
    queued_at: Optional[str] = None

# База данных (просто словарь)
campaign_db: dict[str, dict] = {}

# ЭНДПОИНТЫ МОК-API
@app.post("/mail/send", 
          response_model=MailSendResponse,
          summary="Отправка email-кампании",
          description="Создает кампанию в статусе queued/sending/failed")
async def send_mail(request: MailSendRequest):
    """
    **Параметры запроса:**
    - `campaign_id` (обязательный, str): Уникальный ID кампании
    - `client_id` (обязательный, str): ID клиента  
    - `recipients` (обязательный, array): Список email (1-1000)
    - `subject` (обязательный, str): Тема письма (макс. 200 символов)
    - `priority` (опциональный, enum): low/normal/high
    
    **Возможные статусы ответа:**
    - queued (50%)
    - sending (30%)
    - failed (20%) с ошибкой
    """
    
    # Рандомный статус (реалистичная статистика)
    statuses = ["queued", "sending", "failed"]
    weights = [0.5, 0.3, 0.2]
    status = random.choices(statuses, weights=weights)[0]
    
    error = None
    if status == "failed":
        errors = ["SMTP connection error", "Invalid recipients", "Quota exceeded"]
        error = random.choice(errors)
    
    # Сохраняем в БД
    campaign_db[request.campaign_id] = {
        "client_id": request.client_id,
        "status": status,
        "recipients_count": len(request.recipients),
        "queued_at": " "
    }
    
    return MailSendResponse(
        campaign_id=request.campaign_id,
        client_id=request.client_id,
        status=status,
        error=error,
        queued_at=" "
    )

@app.get("/campaigns/{campaign_id}/status",
         response_model=CampaignStatusResponse,
         summary="Статус кампании",
         description="Возвращает текущий статус по campaign_id")
async def get_campaign_status(campaign_id: str):
    """
    **Путь параметры:**
    - `campaign_id` (path, str): ID кампании
    
    **Возможные статусы:**
    - queued, sending, completed, failed, not_found
    """
    
    campaign = campaign_db.get(campaign_id)
    
    if not campaign:
        raise HTTPException(
            status_code=404,
            detail=CampaignStatusResponse(
                campaign_id=campaign_id,
                status=CampaignStatus.not_found,
                error="Campaign does not exist"
            )
        )
    
    # Симуляция прогресса для sending
    progress = None
    if campaign["status"] == "sending":
        progress = random.randint(10, 90)
    
    return CampaignStatusResponse(
        campaign_id=campaign_id,
        client_id=campaign.get("client_id"),
        status=campaign["status"],
        progress=progress,
        queued_at=campaign.get("queued_at")
    )

# ДОКУМЕНТАЦИЯ API
@app.get("/")
async def api_info():
    """Информация о мок-API"""
    return {
        "endpoints": [
            {
                "method": "POST",
                "path": "/mail/send",
                "description": "Отправка кампании",
                "request": "MailSendRequest",
                "response": "MailSendResponse"
            },
            {
                "method": "GET", 
                "path": "/campaigns/{campaign_id}/status",
                "description": "Статус кампании",
                "response": "CampaignStatusResponse"
            }
        ]
    }