backend/api.py - master - Xton/Case_13_HTML_Newsletter_Builder

backend/api.py
157 строк7 KB
sdankov37
Догрузка файлов
07 май 2026, 18:45
07 май 2026, 18:457620304
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum
import uvicorn
#Наша база данных
from database import db;

app =FastAPI(
    title="API for Project",
    description="API Для Конструктора HTML-рассылок",
)


class CreateSessionRequest(BaseModel): #Запрос на создание сессии
    session_id :Optional[str] = None
class SessionResponse(BaseModel):#Ответ с данными сессии 
    session_id: str
    current_template_name: str
    started_at: str
    canvas_weight: int
    canvas_height: int
class CreateTemplateRequest(BaseModel):# Запорос на создание шаболна
    session_id: str
    name: str
class TemplateResponse(BaseModel):# Запрос на ответ с данными шаблона
    session_id: str
    template_id: str
    name: str
class BlockData(BaseModel): # Ответ на данные по фигуре(блока)
    block_id: str
    template_id: str
    type: str
    content: Optional[str]
    x: int
    y: int
    width: Optional[int]
    height: Optional[int]
    radius: Optional[int]
    color: str
    fill: bool
    font_family: Optional[str]
    font_size: Optional[int]
class BlockResponse(BaseModel):
    """Ответ с данными блока"""
    block_id: str
    template_id: str
    type: str
    content: Optional[str] = None
    x: int
    y: int
    width: Optional[int] = None
    height: Optional[int] = None
    radius: Optional[int] = None
    color: str
    fill: bool
    font_family: Optional[str] = None
    font_size: Optional[int] = None

class CreateCampaignRequest(BaseModel): # Запрос на отправку письмя для компании
    recipient_email: str
class CampaignResponse(BaseModel):# Хапрос на ответ с данными об кампании 
    campaign_id: str
    recipient_email: str
    status: str
    queued_at: str
    error_message: Optional[str] = None
#ЭНДПОИНТЫ ДЛЯ СЕССИЙ
@app.post("/session", response_model=SessionResponse, tags=["Сессии"])#Создание новой сессии для пользователя
async def create_session(request:CreateSessionRequest = None):
    if request and request.session_id:
        session_id = request.session_id
    else:
        session_id = None
    session = db.create_session(session_id)
    return SessionResponse(**session)

@app.get("/sessions/{session_id}", response_model=SessionResponse, tags=["Сессии"])# Получение данных сессии по ID
async def get_session(session_id: str):
    session = db.get_session(session_id)
    if not session:
        raise HTTPException(status_code=404, detail="Сессия не найдена")
    return SessionResponse(**session)   
#ЭНДПОНИТЫ ДЛЯ ШАБЛОНОВ
@app.post("/templates", response_model=TemplateResponse, tags=["Шаблоны"])#Создание нового шаблона в указанной сессии  
async def create_template(request: CreateTemplateRequest):
    # Проверяем, существует ли сессия
    session = db.get_session(request.session_id)
    if not session:
        raise HTTPException(status_code=404, detail="Сессия не найдена")
    
    template_id = db.create_template(request.session_id, request.name)
    template = db.get_template(template_id)
    return TemplateResponse(**template)
@app.get("/sessions/{session_id}/templates", response_model=List[TemplateResponse], tags=["Шаблоны"])#Получить все шаблоны сессии
async def get_templates(session_id: str):
    session = db.get_session(session_id)
    if not session:
        raise HTTPException(status_code=404, detail="Сессия не найдена")
    
    templates = db.get_all_templates(session_id)
    return [TemplateResponse(**t) for t in templates]
@app.get("/templates/{template_id}", response_model=TemplateResponse, tags=["Шаблоны"])#Получить шаблон по ID
async def get_template(template_id: str):
    template = db.get_template(template_id)
    if not template:
        raise HTTPException(status_code=404, detail="Шаблон не найден")
    return TemplateResponse(**template)
#ЭНДПОИНТЫ ДЛЯ БЛОКОВ
@app.post("/templates/{template_id}/blocks", response_model=BlockResponse, tags=["Блоки"]) #Добавить блок в шаблон
async def add_block(template_id: str, block_data: BlockData):
    # Проверяем, существует ли шаблон
    template = db.get_template(template_id)
    if not template:
        raise HTTPException(status_code=404, detail="Шаблон не найден")
    # Преобразуем Pydantic модель в словарь
    block_dict = block_data.dict(exclude_none=True)
    # Добавляем блок
    block_id = db.add_block(template_id, block_dict)
    # Получаем и возвращаем добавленный блок
    blocks = db.get_blocks(template_id)
    new_block = next((b for b in blocks if b['block_id'] == block_id), None)
    if not new_block:
        raise HTTPException(status_code=500, detail="Не удалось создать блок")
    return BlockResponse(**new_block)

@app.get("/templates/{template_id}/blocks", response_model=List[BlockResponse], tags=["Блоки"])#Получить все блоки шаблона
async def get_blocks(template_id: str):
    template = db.get_template(template_id)
    if not template:
        raise HTTPException(status_code=404, detail="Шаблон не найден") 
    blocks = db.get_blocks(template_id)
    return [BlockResponse(**b) for b in blocks]
#ЭНДПОИНТ ДЛЯ ОТПРАВКИ ПИСЕМ 
@app.post("/campaigns", response_model=CampaignResponse, tags=["Кампании"])#Отправка письма 
async def create_campaign(request: CreateCampaignRequest):
    result = db.create_campaign(request.recipient_email)
    return CampaignResponse(
        campaign_id=result['campaign_id'],
        recipient_email=request.recipient_email,
        status=result['status'],
        queued_at=result['queued_at'],
        error_message=result.get('error_message')
    )
@app.get("/campaigns/{campaign_id}/status", response_model=CampaignResponse, tags=["Кампании"])#Статусы компаний(failed, completed, queued, sending)
async def get_campaign_status(campaign_id: str):
    campaign = db.get_campaign_status(campaign_id)
    if not campaign:
        raise HTTPException(status_code=404, detail="Кампания не найдена")
    return CampaignResponse(**campaign)

@app.get("/campaigns", response_model=List[CampaignResponse], tags=["Кампании"])#Получить все кампании
async def get_all_campaigns():
    campaigns = db.get_all_campaigns()
    return [CampaignResponse(**c) for c in campaigns]
if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=8000)