FastAPI Quickstart

pip install fastapi uvicorn sendpost-python-sdk python-dotenv pydantic

from fastapi import FastAPI
from dotenv import load_dotenv
import os

load_dotenv()

app = FastAPI(title="SendPost Email API")

@app.get("/")
def read_root():
    return {"message": "SendPost Email API"}

import sendpost_python_sdk
from sendpost_python_sdk.api import EmailApi
from sendpost_python_sdk.models import EmailMessageObject, EmailAddress, Recipient
import os
import re
from typing import List, Dict, Any, Optional, Union

class SendPostEmailService:
    def __init__(self):
        self.api_key = os.getenv('SENDPOST_API_KEY')
        self.from_email = os.getenv('SENDPOST_FROM_EMAIL')
        self.from_name = os.getenv('SENDPOST_FROM_NAME')
        self.configuration = sendpost_python_sdk.Configuration(
            host="https://api.sendpost.io/api/v1"
        )
        self.configuration.api_key['subAccountAuth'] = self.api_key
    
    async def send_email(
        self,
        to: Union[str, List[Union[str, Dict[str, Any]]]],
        subject: str,
        html_body: str,
        text_body: Optional[str] = None,
        from_email: Optional[str] = None,
        from_name: Optional[str] = None,
        groups: Optional[List[str]] = None,
        track_opens: bool = True,
        track_clicks: bool = True,
        custom_fields: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """
        Send an email using SendPost API.
        
        Args:
            to: Email address (str) or list of Recipient objects/dicts
            subject: Email subject
            html_body: HTML email body
            text_body: Plain text email body (optional)
            from_email: Sender email (optional)
            from_name: Sender name (optional)
            groups: List of groups for email classification
            track_opens: Enable open tracking
            track_clicks: Enable click tracking
            custom_fields: Dict of custom fields for personalization
        
        Returns:
            dict: {'success': bool, 'message_id': str or None, 'error': str or None}
        """
        try:
            with sendpost_python_sdk.ApiClient(self.configuration) as api_client:
                email_message = EmailMessageObject()
                
                # Set sender
                email_message.var_from = EmailAddress(
                    email=from_email or self.from_email,
                    name=from_name or self.from_name
                )
                
                # Handle recipients
                if isinstance(to, str):
                    recipients = [Recipient(email=to)]
                elif isinstance(to, list):
                    recipients = []
                    for recipient in to:
                        if isinstance(recipient, str):
                            recipients.append(Recipient(email=recipient))
                        elif isinstance(recipient, dict):
                            recipients.append(Recipient(
                                email=recipient.get('email'),
                                name=recipient.get('name'),
                                custom_fields=recipient.get('customFields') or custom_fields
                            ))
                        else:
                            recipients.append(recipient)
                else:
                    recipients = [Recipient(email=to)]
                
                email_message.to = recipients
                email_message.subject = subject
                email_message.html_body = html_body
                email_message.text_body = text_body or self._html_to_text(html_body)
                email_message.track_opens = track_opens
                email_message.track_clicks = track_clicks
                
                if groups:
                    email_message.groups = groups
                
                response = EmailApi(api_client).send_email(email_message)[0]
                
                return {
                    'success': True,
                    'message_id': response.message_id,
                    'data': response
                }
        except sendpost_python_sdk.exceptions.ApiException as e:
            return {
                'success': False,
                'message_id': None,
                'error': f"API Error {e.status}: {e.body}"
            }
        except Exception as e:
            return {
                'success': False,
                'message_id': None,
                'error': str(e)
            }
    
    @staticmethod
    def _html_to_text(html: str) -> str:
        """Simple HTML to text conversion"""
        text = re.sub(r'<[^>]+>', '', html)
        text = text.replace('&nbsp;', ' ')
        text = text.replace('&amp;', '&')
        text = text.replace('&lt;', '<')
        text = text.replace('&gt;', '>')
        return text.strip()

# Singleton instance
email_service = SendPostEmailService()

from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional, Dict, Any, Union

class EmailRecipient(BaseModel):
    email: EmailStr
    name: Optional[str] = None
    customFields: Optional[Dict[str, Any]] = None

class SendEmailRequest(BaseModel):
    to: Union[EmailStr, List[EmailRecipient]]
    subject: str = Field(..., min_length=1)
    htmlBody: str = Field(..., min_length=1)
    textBody: Optional[str] = None
    from_email: Optional[EmailStr] = Field(None, alias='from')
    from_name: Optional[str] = None
    groups: Optional[List[str]] = None
    trackOpens: bool = True
    trackClicks: bool = True

class EmailResponse(BaseModel):
    success: bool
    messageId: Optional[str] = None
    error: Optional[str] = None

class WelcomeEmailRequest(BaseModel):
    email: EmailStr
    firstName: str = Field(..., min_length=1)

class PasswordResetRequest(BaseModel):
    email: EmailStr
    resetToken: str = Field(..., min_length=1)
    firstName: Optional[str] = 'User'

from fastapi import FastAPI, HTTPException
from models.email import SendEmailRequest, EmailResponse
from services.email_service import email_service

app = FastAPI()

@app.post("/api/send-email", response_model=EmailResponse)
async def send_email(request: SendEmailRequest):
    """Send an email using SendPost"""
    result = await email_service.send_email(
        to=request.to,
        subject=request.subject,
        html_body=request.htmlBody,
        text_body=request.textBody,
        from_email=request.from_email,
        from_name=request.from_name,
        groups=request.groups,
        track_opens=request.trackOpens,
        track_clicks=request.trackClicks
    )
    
    if result['success']:
        return EmailResponse(
            success=True,
            messageId=result['message_id']
        )
    else:
        raise HTTPException(
            status_code=500,
            detail=result['error']
        )

from models.email import WelcomeEmailRequest, EmailResponse

@app.post("/api/welcome-email", response_model=EmailResponse)
async def send_welcome_email(request: WelcomeEmailRequest):
    """Send welcome email to new user"""
    result = await email_service.send_email(
        to={
            'email': request.email,
            'name': request.firstName,
            'customFields': {'firstName': request.firstName}
        },
        subject='Welcome {{firstName}}!',
        html_body='''
            <h1>Welcome, {{firstName}}!</h1>
            <p>Thank you for joining us. We're excited to have you on board.</p>
            <p>Get started by exploring our features:</p>
            <ul>
                <li>Feature 1</li>
                <li>Feature 2</li>
                <li>Feature 3</li>
            </ul>
            <p>If you have any questions, just reply to this email.</p>
            <p><a href="{{unsubscribe}}">Unsubscribe</a></p>
        ''',
        text_body='''
            Welcome, {{firstName}}!
            
            Thank you for joining us. We're excited to have you on board.
            
            Get started by exploring our features:
            - Feature 1
            - Feature 2
            - Feature 3
            
            If you have any questions, just reply to this email.
            
            Unsubscribe: {{unsubscribe}}
        ''',
        groups=['welcome']
    )
    
    if result['success']:
        return EmailResponse(
            success=True,
            messageId=result['message_id']
        )
    else:
        raise HTTPException(
            status_code=500,
            detail=result['error']
        )

from models.email import PasswordResetRequest
from fastapi import Request

@app.post("/api/password-reset", response_model=EmailResponse)
async def send_password_reset(request: Request, data: PasswordResetRequest):
    """Send password reset email"""
    base_url = str(request.base_url).rstrip('/')
    reset_url = f"{base_url}/reset-password?token={data.resetToken}"
    
    result = await email_service.send_email(
        to={
            'email': data.email,
            'name': data.firstName,
            'customFields': {
                'firstName': data.firstName,
                'resetUrl': reset_url
            }
        },
        subject='Reset Your Password',
        html_body=f'''
            <h2>Hello {{firstName}},</h2>
            <p>We received a request to reset your password. Click the button below:</p>
            <p>
                <a href="{{resetUrl}}" style="background-color: #5750EC; color: white; padding: 12px 24px; text-decoration: none; border-radius: 4px; display: inline-block;">
                    Reset Password
                </a>
            </p>
            <p>Or copy and paste this link: {{resetUrl}}</p>
            <p>This link will expire in 1 hour.</p>
            <p>If you didn't request this, please ignore this email.</p>
        ''',
        text_body='''
            Hello {{firstName}},
            
            We received a request to reset your password. Click the link below:
            
            {{resetUrl}}
            
            This link will expire in 1 hour.
            
            If you didn't request this, please ignore this email.
        ''',
        groups=['password-reset'],
        track_clicks=True
    )
    
    if result['success']:
        return EmailResponse(
            success=True,
            messageId=result['message_id']
        )
    else:
        raise HTTPException(
            status_code=500,
            detail=result['error']
        )

from fastapi import Depends
from services.email_service import SendPostEmailService

def get_email_service() -> SendPostEmailService:
    """Dependency to get email service instance"""
    return SendPostEmailService()

# Or use a singleton
from services.email_service import email_service

def get_email_service() -> SendPostEmailService:
    return email_service

from dependencies import get_email_service

@app.post("/api/send-email")
async def send_email(
    request: SendEmailRequest,
    email_service: SendPostEmailService = Depends(get_email_service)
):
    result = await email_service.send_email(
        to=request.to,
        subject=request.subject,
        html_body=request.htmlBody
    )
    # ... rest of the code

from fastapi import BackgroundTasks

@app.post("/api/send-email-async")
async def send_email_async(
    request: SendEmailRequest,
    background_tasks: BackgroundTasks
):
    """Queue email to be sent in background"""
    background_tasks.add_task(
        email_service.send_email,
        to=request.to,
        subject=request.subject,
        html_body=request.htmlBody,
        text_body=request.textBody
    )
    
    return {"success": True, "message": "Email queued for sending"}

from fastapi import HTTPException

class EmailSendException(HTTPException):
    def __init__(self, detail: str):
        super().__init__(status_code=500, detail=detail)

class InvalidEmailException(HTTPException):
    def __init__(self, detail: str = "Invalid email address"):
        super().__init__(status_code=400, detail=detail)

from exceptions import EmailSendException

@app.post("/api/send-email")
async def send_email(request: SendEmailRequest):
    result = await email_service.send_email(
        to=request.to,
        subject=request.subject,
        html_body=request.htmlBody
    )
    
    if not result['success']:
        raise EmailSendException(result['error'])
    
    return EmailResponse(
        success=True,
        messageId=result['message_id']
    )

uvicorn main:app --reload