FastAPI

myapp/
├── main.py             # FastAPI application
├── config.py           # Configuration settings
├── models/
│   └── email.py        # Pydantic models
├── services/
│   └── email.py        # Email service
├── routers/
│   ├── __init__.py
│   ├── auth.py         # Authentication routes
│   └── api.py          # API routes
├── .env                # Environment variables
└── requirements.txt    # Dependencies

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings."""
    lettr_api_key: str
    mail_from_address: str = "notifications@yourdomain.com"
    mail_from_name: str = "My Application"
    environment: str = "development"

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    """Get cached settings instance."""
    return Settings()

from pydantic import BaseModel, EmailStr
from typing import List, Optional, Dict

class EmailRequest(BaseModel):
    """Email sending request."""
    to: List[EmailStr]
    subject: str
    html: str
    from_email: Optional[EmailStr] = None
    metadata: Optional[Dict[str, str]] = None

class TemplateEmailRequest(BaseModel):
    """Template email sending request."""
    to: List[EmailStr]
    template_slug: str
    substitution_data: Dict[str, str]

import lettr
from config import get_settings

settings = get_settings()

class EmailService:
    """Async email service for sending transactional emails."""

    def __init__(self):
        self.client = lettr.AsyncLettr(settings.lettr_api_key)

    async def send_welcome_email(self, recipient: str, name: str) -> str:
        """Send a welcome email to a new user."""
        response = await self.client.emails.send(
            from_email=settings.mail_from_address,
            from_name=settings.mail_from_name,
            to=[recipient],
            subject=f"Welcome to {settings.mail_from_name}, {name}!",
            html=f"""
                <h1>Welcome, {name}!</h1>
                <p>Thanks for signing up. We're excited to have you on board.</p>
            """,
            metadata={"email_type": "welcome", "user_email": recipient},
        )
        return response.request_id

    async def send_password_reset(self, recipient: str, reset_token: str) -> str:
        """Send a password reset email."""
        reset_url = f"https://yourdomain.com/reset-password?token={reset_token}"
        response = await self.client.emails.send(
            from_email=settings.mail_from_address,
            to=[recipient],
            subject="Reset your password",
            html=f"""
                <h1>Reset your password</h1>
                <p><a href="{reset_url}">Reset Password</a></p>
                <p>This link expires in 1 hour.</p>
            """,
        )
        return response.request_id

# Dependency injection
async def get_email_service() -> EmailService:
    """Get email service instance."""
    return EmailService()

from fastapi import APIRouter, HTTPException, Depends, status
from pydantic import BaseModel, EmailStr
from services.email import EmailService, get_email_service
import logging

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/auth", tags=["authentication"])

class RegisterRequest(BaseModel):
    email: EmailStr
    name: str
    password: str

@router.post("/register", status_code=status.HTTP_201_CREATED)
async def register(
    request: RegisterRequest,
    email_service: EmailService = Depends(get_email_service),
):
    """Register a new user and send welcome email."""
    # ... validate data and create user ...

    try:
        request_id = await email_service.send_welcome_email(request.email, request.name)
        return {"success": True, "email_request_id": request_id}
    except Exception as e:
        logger.error(f"Failed to send welcome email: {e}")
        return {"success": True, "message": "Registration successful, but email failed to send"}

@router.post("/forgot-password")
async def forgot_password(
    email: EmailStr,
    email_service: EmailService = Depends(get_email_service),
):
    """Send password reset email."""
    reset_token = "example_token_123"  # ... generate reset token ...
    request_id = await email_service.send_password_reset(email, reset_token)
    return {"success": True, "request_id": request_id}

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routers import auth, api
from config import get_settings
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

settings = get_settings()

app = FastAPI(
    title="My API",
    description="API with Lettr email integration",
    version="1.0.0",
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
app.include_router(auth.router)
app.include_router(api.router)

@app.get("/")
async def root():
    """Root endpoint."""
    return {"message": "API is running"}

@app.on_event("startup")
async def startup_event():
    """Startup event handler."""
    logger.info("Application starting up...")

@app.on_event("shutdown")
async def shutdown_event():
    """Shutdown event handler."""
    logger.info("Application shutting down...")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from models.email import TemplateEmailRequest

@app.post("/send-template")
async def send_template(
    request: TemplateEmailRequest,
    email_service: EmailService = Depends(get_email_service),
):
    """Send an email using a template."""
    response = await email_service.client.emails.send(
        from_email=settings.mail_from_address,
        to=request.to,
        template_slug=request.template_slug,
        substitution_data=request.substitution_data,
    )
    return {"success": True, "request_id": response.request_id}

from fastapi import BackgroundTasks

async def send_email_background(recipient: str, subject: str, html: str):
    """Background task for sending emails."""
    email_service = EmailService()
    await email_service.client.emails.send(
        from_email=settings.mail_from_address,
        to=[recipient],
        subject=subject,
        html=html,
    )

@app.post("/register")
async def register(
    email: EmailStr,
    name: str,
    background_tasks: BackgroundTasks,
):
    """Register user and send welcome email in background."""
    # ... create user ...

    # Add email to background tasks
    background_tasks.add_task(
        send_email_background,
        email,
        "Welcome!",
        f"<h1>Welcome, {name}!</h1>",
    )

    return {"success": True, "message": "Registration successful"}

import asyncio
from typing import List

class BatchEmailRequest(BaseModel):
    recipients: List[EmailStr]
    subject: str
    html: str

@app.post("/send-batch")
async def send_batch(
    request: BatchEmailRequest,
    email_service: EmailService = Depends(get_email_service),
):
    """Send emails to multiple recipients concurrently."""
    async def send_to_recipient(recipient: str):
        return await email_service.client.emails.send(
            from_email=settings.mail_from_address,
            to=[recipient],
            subject=request.subject,
            html=request.html,
        )

    results = await asyncio.gather(
        *[send_to_recipient(r) for r in request.recipients],
        return_exceptions=True,
    )

    successful = sum(1 for r in results if not isinstance(r, Exception))
    return {"total": len(results), "successful": successful, "failed": len(results) - successful}

from fastapi import HTTPException, status
import lettr

@app.post("/send-email")
async def send_email(request: EmailRequest):
    try:
        response = await client.emails.send(...)
        return {"success": True, "request_id": response.request_id}
    except lettr.ValidationError as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="Invalid email parameters"
        )
    except lettr.LettrError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Email service error"
        )

pip install pytest pytest-asyncio httpx

import pytest
from httpx import AsyncClient
from main import app
from unittest.mock import Mock, patch

@pytest.mark.asyncio
async def test_register_endpoint():
    """Test user registration endpoint."""
    async with AsyncClient(app=app, base_url="http://test") as client:
        with patch("services.email.EmailService.send_welcome_email") as mock_send:
            mock_send.return_value = "test-request-id"

            response = await client.post("/auth/register", json={
                "email": "test@example.com",
                "name": "Test User",
                "password": "password123",
            })

            assert response.status_code == 201
            assert response.json()["success"] is True

from pydantic import BaseModel, EmailStr, validator

class EmailRequest(BaseModel):
    to: EmailStr
    subject: str
    html: str

    @validator("subject")
    def subject_not_empty(cls, v):
        if not v.strip():
            raise ValueError("Subject cannot be empty")
        return v

import logging

logger = logging.getLogger(__name__)

@app.post("/send-email")
async def send_email(request: EmailRequest):
    response = await client.emails.send(...)
    logger.info(
        "Email sent successfully",
        extra={"request_id": response.request_id, "to": request.to}
    )
    return {"success": True}

pip install slowapi

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/send-email")
@limiter.limit("10/minute")
async def send_email(request: Request, email: EmailRequest):
    # ... send email ...
    pass