Files
orion/app/api/v1/vendor/email_settings.py
Samir Boulahtit 84a523cd7b fix: resolve email settings architecture violations and add tests/docs
- Fix API-002 in admin/settings.py: use service layer for DB delete
- Fix API-001/API-003 in vendor/email_settings.py: add Pydantic response
  models, remove HTTPException raises
- Fix SVC-002/SVC-006 in vendor_email_settings_service.py: use domain
  exceptions, change db.commit() to db.flush()
- Add unit tests for VendorEmailSettingsService
- Add integration tests for vendor and admin email settings APIs
- Add user guide (docs/guides/email-settings.md)
- Add developer guide (docs/implementation/email-settings.md)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-05 22:38:10 +01:00

277 lines
8.4 KiB
Python

# app/api/v1/vendor/email_settings.py
"""
Vendor email settings API endpoints.
Allows vendors to configure their email sending settings:
- SMTP configuration (all tiers)
- Advanced providers: SendGrid, Mailgun, SES (Business+ tier)
- Sender identity (from_email, from_name, reply_to)
- Signature/footer customization
- Configuration verification via test email
Vendor Context: Uses token_vendor_id from JWT token (authenticated vendor API pattern).
"""
import logging
from fastapi import APIRouter, Depends
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy.orm import Session
from app.api.deps import get_current_vendor_api
from app.core.database import get_db
from app.services.vendor_email_settings_service import VendorEmailSettingsService
from app.services.subscription_service import subscription_service
from models.database.user import User
router = APIRouter(prefix="/email-settings")
logger = logging.getLogger(__name__)
# =============================================================================
# SCHEMAS
# =============================================================================
class EmailSettingsUpdate(BaseModel):
"""Schema for creating/updating email settings."""
# Sender Identity (Required)
from_email: EmailStr = Field(..., description="Sender email address")
from_name: str = Field(..., min_length=1, max_length=100, description="Sender name")
reply_to_email: EmailStr | None = Field(None, description="Reply-to email address")
# Signature (Optional)
signature_text: str | None = Field(None, description="Plain text signature")
signature_html: str | None = Field(None, description="HTML signature/footer")
# Provider
provider: str = Field("smtp", description="Email provider: smtp, sendgrid, mailgun, ses")
# SMTP Settings
smtp_host: str | None = Field(None, description="SMTP server hostname")
smtp_port: int | None = Field(587, ge=1, le=65535, description="SMTP server port")
smtp_username: str | None = Field(None, description="SMTP username")
smtp_password: str | None = Field(None, description="SMTP password")
smtp_use_tls: bool = Field(True, description="Use STARTTLS")
smtp_use_ssl: bool = Field(False, description="Use SSL/TLS (port 465)")
# SendGrid
sendgrid_api_key: str | None = Field(None, description="SendGrid API key")
# Mailgun
mailgun_api_key: str | None = Field(None, description="Mailgun API key")
mailgun_domain: str | None = Field(None, description="Mailgun sending domain")
# SES
ses_access_key_id: str | None = Field(None, description="AWS access key ID")
ses_secret_access_key: str | None = Field(None, description="AWS secret access key")
ses_region: str | None = Field("eu-west-1", description="AWS region")
class VerifyEmailRequest(BaseModel):
"""Schema for verifying email settings."""
test_email: EmailStr = Field(..., description="Email address to send test email to")
# Response models for API-001 compliance
class EmailSettingsResponse(BaseModel):
"""Response for email settings."""
configured: bool
verified: bool | None = None
settings: dict | None = None
message: str | None = None
class EmailStatusResponse(BaseModel):
"""Response for email status check."""
is_configured: bool
is_verified: bool
class ProvidersResponse(BaseModel):
"""Response for available providers."""
providers: list[dict]
current_tier: str | None
class EmailUpdateResponse(BaseModel):
"""Response for email settings update."""
success: bool
message: str
settings: dict
class EmailVerifyResponse(BaseModel):
"""Response for email verification."""
success: bool
message: str
class EmailDeleteResponse(BaseModel):
"""Response for email settings deletion."""
success: bool
message: str
# =============================================================================
# ENDPOINTS
# =============================================================================
@router.get("", response_model=EmailSettingsResponse)
def get_email_settings(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
) -> EmailSettingsResponse:
"""
Get current email settings for the vendor.
Returns settings with sensitive fields masked.
"""
vendor_id = current_user.token_vendor_id
service = VendorEmailSettingsService(db)
settings = service.get_settings(vendor_id)
if not settings:
return EmailSettingsResponse(
configured=False,
settings=None,
message="Email settings not configured. Configure SMTP to send emails to customers.",
)
return EmailSettingsResponse(
configured=settings.is_configured,
verified=settings.is_verified,
settings=settings.to_dict(),
)
@router.get("/status", response_model=EmailStatusResponse)
def get_email_status(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
) -> EmailStatusResponse:
"""
Get email configuration status.
Used by frontend to show warning banner if not configured.
"""
vendor_id = current_user.token_vendor_id
service = VendorEmailSettingsService(db)
status = service.get_status(vendor_id)
return EmailStatusResponse(**status)
@router.get("/providers", response_model=ProvidersResponse)
def get_available_providers(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
) -> ProvidersResponse:
"""
Get available email providers for current tier.
Returns list of providers with availability status.
"""
vendor_id = current_user.token_vendor_id
service = VendorEmailSettingsService(db)
# Get vendor's current tier
tier = subscription_service.get_current_tier(db, vendor_id)
return ProvidersResponse(
providers=service.get_available_providers(tier),
current_tier=tier.value if tier else None,
)
@router.put("", response_model=EmailUpdateResponse)
def update_email_settings(
data: EmailSettingsUpdate,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
) -> EmailUpdateResponse:
"""
Create or update email settings.
Premium providers (SendGrid, Mailgun, SES) require Business+ tier.
Raises AuthorizationException if tier is insufficient.
Raises ValidationException if data is invalid.
"""
vendor_id = current_user.token_vendor_id
service = VendorEmailSettingsService(db)
# Get vendor's current tier for validation
tier = subscription_service.get_current_tier(db, vendor_id)
# Service raises appropriate exceptions (API-003 compliance)
settings = service.create_or_update(
vendor_id=vendor_id,
data=data.model_dump(exclude_unset=True),
current_tier=tier,
)
db.commit()
return EmailUpdateResponse(
success=True,
message="Email settings updated successfully",
settings=settings.to_dict(),
)
@router.post("/verify", response_model=EmailVerifyResponse)
def verify_email_settings(
data: VerifyEmailRequest,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
) -> EmailVerifyResponse:
"""
Verify email settings by sending a test email.
Sends a test email to the provided address and updates verification status.
Raises ResourceNotFoundException if settings not configured.
Raises ValidationException if verification fails.
"""
vendor_id = current_user.token_vendor_id
service = VendorEmailSettingsService(db)
# Service raises appropriate exceptions (API-003 compliance)
result = service.verify_settings(vendor_id, data.test_email)
db.commit()
return EmailVerifyResponse(
success=result["success"],
message=result["message"],
)
@router.delete("", response_model=EmailDeleteResponse)
def delete_email_settings(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
) -> EmailDeleteResponse:
"""
Delete email settings.
Warning: This will disable email sending for the vendor.
Raises ResourceNotFoundException if settings not found.
"""
vendor_id = current_user.token_vendor_id
service = VendorEmailSettingsService(db)
# Service raises ResourceNotFoundException if not found (API-003 compliance)
service.delete(vendor_id)
db.commit()
return EmailDeleteResponse(
success=True,
message="Email settings deleted",
)