Cheng Zhou
77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.deps import get_db_session, require_roles
|
|
from app.schemas.email_settings import (
|
|
EmailDomainCreateRequest,
|
|
EmailSettingsListResponse,
|
|
EmailSettingsRead,
|
|
EmailSettingsUpdate,
|
|
EmailTestRequest,
|
|
)
|
|
from app.services import email_service
|
|
|
|
router = APIRouter(prefix="/email-settings")
|
|
|
|
|
|
@router.get("/", response_model=EmailSettingsListResponse)
|
|
async def read_email_settings(
|
|
db: AsyncSession = Depends(get_db_session),
|
|
current_user=Depends(require_roles(["ADMIN"])),
|
|
) -> EmailSettingsListResponse:
|
|
items = [email_service.to_email_settings_read(row) for row in await email_service.list_email_settings(db)]
|
|
return EmailSettingsListResponse(items=items, total=len(items))
|
|
|
|
|
|
@router.post("/", response_model=EmailSettingsRead)
|
|
async def create_email_domain(
|
|
payload: EmailDomainCreateRequest,
|
|
db: AsyncSession = Depends(get_db_session),
|
|
current_user=Depends(require_roles(["ADMIN"])),
|
|
) -> EmailSettingsRead:
|
|
settings_row = await email_service.create_email_domain(db, payload.register_domain, updated_by=current_user.id)
|
|
return email_service.to_email_settings_read(settings_row)
|
|
|
|
|
|
@router.get("/{register_domain}", response_model=EmailSettingsRead)
|
|
async def read_email_domain_settings(
|
|
register_domain: str,
|
|
db: AsyncSession = Depends(get_db_session),
|
|
current_user=Depends(require_roles(["ADMIN"])),
|
|
) -> EmailSettingsRead:
|
|
return email_service.to_email_settings_read(await email_service.get_email_settings(db, register_domain))
|
|
|
|
|
|
@router.put("/{register_domain}", response_model=EmailSettingsRead)
|
|
async def update_email_settings(
|
|
register_domain: str,
|
|
payload: EmailSettingsUpdate,
|
|
db: AsyncSession = Depends(get_db_session),
|
|
current_user=Depends(require_roles(["ADMIN"])),
|
|
) -> EmailSettingsRead:
|
|
settings_row = await email_service.upsert_email_settings(db, register_domain, payload, updated_by=current_user.id)
|
|
return email_service.to_email_settings_read(settings_row)
|
|
|
|
|
|
@router.delete("/{register_domain}")
|
|
async def delete_email_domain(
|
|
register_domain: str,
|
|
db: AsyncSession = Depends(get_db_session),
|
|
current_user=Depends(require_roles(["ADMIN"])),
|
|
):
|
|
await email_service.delete_email_domain(db, register_domain)
|
|
return {"message": "邮箱后缀已删除"}
|
|
|
|
|
|
@router.post("/{register_domain}/test")
|
|
async def test_email_settings(
|
|
register_domain: str,
|
|
payload: EmailTestRequest,
|
|
db: AsyncSession = Depends(get_db_session),
|
|
current_user=Depends(require_roles(["ADMIN"])),
|
|
):
|
|
await email_service.send_test_email(db, register_domain, str(payload.recipient_email))
|
|
return {"message": "测试邮件已发送"}
|