Flask

This guide shows you how to integrate Unsent into your Flask application for sending transactional emails, managing contacts, and handling webhooks.

Prerequisites

Python 3.7+: Flask requires Python 3.7 or higher

Flask 2.0+: This guide assumes Flask 2.0 or newer

Unsent API Key: Generate one in your Unsent dashboard

Verified Domain: Set up a domain in the Domains section

Installation

Install Flask and the Unsent SDK:

pip install flask unsent

Configuration

Environment Variables

Create a .env file for your configuration:

.env

UNSENT_API_KEY=un_...
UNSENT_FROM_EMAIL=noreply@yourdomain.com
FLASK_APP=app.py
FLASK_ENV=development

Never commit your .env file to version control. Add it to .gitignore and use environment variables in production.

Flask App Setup

Create your Flask application with Unsent configured:

app.py

import os
from flask import Flask
from unsent import unsent
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
app.config['UNSENT_API_KEY'] = os.getenv('UNSENT_API_KEY')
app.config['UNSENT_FROM_EMAIL'] = os.getenv('UNSENT_FROM_EMAIL')

# Create a global Unsent client
unsent_client = unsent(app.config['UNSENT_API_KEY'], raise_on_error=False)

@app.route('/')
def index():
    return 'Flask + Unsent Email Service'

if __name__ == '__main__':
    app.run(debug=True)

Quick Start

Sending a Simple Email

Create an endpoint to send emails:

app.py

from flask import Flask, request, jsonify
from unsent import unsent, types
import os

app = Flask(__name__)
client = unsent(os.getenv('UNSENT_API_KEY'), raise_on_error=False)

@app.route('/send-email', methods=['POST'])
def send_email():
    data = request.get_json()
    
    payload: types.EmailCreate = {
        "to": data['to'],
        "from": os.getenv('UNSENT_FROM_EMAIL'),
        "subject": data['subject'],
        "html": f"<h1>{data['message']}</h1>",
        "text": data['message']
    }
    
    result, error = client.emails.send(payload)
    
    if error:
        app.logger.error(f"Failed to send email: {error}")
        return jsonify({"error": error}), 500
    
    return jsonify({"success": True, "email_id": result.get('id')}), 200

Test it with curl:

curl -X POST http://localhost:5000/send-email \
  -H "Content-Type: application/json" \
  -d '{"to":"user@example.com","subject":"Hello","message":"Welcome!"}'

Blueprint Organization

Structure your email functionality using Blueprints:

Create Email Service

services/email_service.py

from unsent import unsent, types
from typing import Optional, Tuple, Dict, Any
import os

class EmailService:
    def __init__(self):
        self.client = unsent(os.getenv('UNSENT_API_KEY'), raise_on_error=False)
        self.from_email = os.getenv('UNSENT_FROM_EMAIL')
    
    def send_welcome_email(self, user_email: str, user_name: str) -> Tuple[Optional[Dict], Optional[Dict]]:
        """Send a welcome email to a new user."""
        payload: types.EmailCreate = {
            "to": user_email,
            "from": self.from_email,
            "subject": "Welcome to Our App!",
            "html": f"""
                <h1>Welcome, {user_name}!</h1>
                <p>Thanks for joining us. We're excited to have you on board.</p>
                <p>Here's what you can do next:</p>
                <ul>
                    <li>Complete your profile</li>
                    <li>Explore our features</li>
                    <li>Connect with other users</li>
                </ul>
            """,
            "text": f"Welcome, {user_name}! Thanks for joining us."
        }
        
        return self.client.emails.send(payload)
    
    def send_password_reset(self, user_email: str, reset_token: str) -> Tuple[Optional[Dict], Optional[Dict]]:
        """Send a password reset email."""
        reset_url = f"{os.getenv('APP_URL')}/reset-password?token={reset_token}"
        
        payload: types.EmailCreate = {
            "to": user_email,
            "from": self.from_email,
            "subject": "Reset Your Password",
            "html": f"""
                <h2>Password Reset Request</h2>
                <p>Click the button below to reset your password:</p>
                <a href="{reset_url}" style="display: inline-block; padding: 12px 24px; background-color: #0070f3; color: white; text-decoration: none; border-radius: 6px;">Reset Password</a>
                <p>Or copy this link: {reset_url}</p>
                <p>This link will expire in 1 hour.</p>
            """,
            "text": f"Reset your password: {reset_url}"
        }
        
        return self.client.emails.send(payload)

# Create a singleton instance
email_service = EmailService()

Create Blueprint

routes/auth.py

from flask import Blueprint, request, jsonify
from services.email_service import email_service
import secrets

auth_bp = Blueprint('auth', __name__, url_prefix='/auth')

@auth_bp.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    
    # Create user in database (simplified)
    user_email = data['email']
    user_name = data['name']
    
    # Send welcome email
    result, error = email_service.send_welcome_email(user_email, user_name)
    
    if error:
        return jsonify({"error": "Failed to send welcome email"}), 500
    
    return jsonify({
        "message": "User registered successfully",
        "email_sent": True
    }), 201

@auth_bp.route('/forgot-password', methods=['POST'])
def forgot_password():
    data = request.get_json()
    user_email = data['email']
    
    # Generate reset token (in production, store this in DB)
    reset_token = secrets.token_urlsafe(32)
    
    # Send reset email
    result, error = email_service.send_password_reset(user_email, reset_token)
    
    if error:
        return jsonify({"error": "Failed to send reset email"}), 500
    
    return jsonify({"message": "Password reset email sent"}), 200

Register Blueprint

app.py

from flask import Flask
from routes.auth import auth_bp

app = Flask(__name__)
app.register_blueprint(auth_bp)

if __name__ == '__main__':
    app.run(debug=True)

Jinja2 Email Templates

Use Jinja2 templates for better email management:

Create Templates

templates/emails/welcome.html

<!DOCTYPE html>
<html>
<head>
    <style>
        body { font-family: Arial, sans-serif; line-height: 1.6; color: #333; }
        .container { max-width: 600px; margin: 0 auto; padding: 20px; }
        .header { background-color: #0070f3; color: white; padding: 20px; text-align: center; }
        .content { padding: 20px; background-color: #f9f9f9; }
        .button { display: inline-block; padding: 12px 24px; background-color: #0070f3; color: white; text-decoration: none; border-radius: 6px; }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>Welcome to {{ app_name }}!</h1>
        </div>
        <div class="content">
            <p>Hi {{ user_name }},</p>
            <p>Thanks for joining {{ app_name }}! We're excited to have you on board.</p>
            <p>Here's what you can do next:</p>
            <ul>
                <li>Complete your profile</li>
                <li>Explore our features</li>
                <li>Connect with other users</li>
            </ul>
            <p style="text-align: center; margin-top: 30px;">
                <a href="{{ app_url }}" class="button">Get Started</a>
            </p>
        </div>
    </div>
</body>
</html>

Render Templates

services/email_service.py

from flask import current_app, render_template
from unsent import unsent, types
import os

class EmailService:
    def __init__(self):
        self.client = unsent(os.getenv('UNSENT_API_KEY'), raise_on_error=False)
        self.from_email = os.getenv('UNSENT_FROM_EMAIL')
    
    def send_welcome_email(self, user_email: str, user_name: str):
        """Send a welcome email using Jinja2 template."""
        html = render_template(
            'emails/welcome.html',
            user_name=user_name,
            app_name=os.getenv('APP_NAME', 'Our App'),
            app_url=os.getenv('APP_URL')
        )
        
        payload: types.EmailCreate = {
            "to": user_email,
            "from": self.from_email,
            "subject": f"Welcome to {os.getenv('APP_NAME')}!",
            "html": html,
            "text": f"Welcome, {user_name}! Thanks for joining {os.getenv('APP_NAME')}."
        }
        
        return self.client.emails.send(payload)

Background Jobs with Celery

Send emails asynchronously using Celery:

Install Celery

pip install celery redis

Configure Celery

celery_app.py

from celery import Celery
import os

def make_celery(app_name=__name__):
    redis_url = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
    
    celery_app = Celery(
        app_name,
        broker=redis_url,
        backend=redis_url
    )
    
    celery_app.conf.update(
        task_serializer='json',
        accept_content=['json'],
        result_serializer='json',
        timezone='UTC',
        enable_utc=True,
    )
    
    return celery_app

celery = make_celery()

Create Email Tasks

tasks/email_tasks.py

from celery_app import celery
from unsent import unsent, types
import os

client = unsent(os.getenv('UNSENT_API_KEY'), raise_on_error=False)

@celery.task(name='tasks.send_welcome_email', bind=True, max_retries=3)
def send_welcome_email(self, user_email: str, user_name: str):
    """Background task to send welcome email."""
    try:
        payload: types.EmailCreate = {
            "to": user_email,
            "from": os.getenv('UNSENT_FROM_EMAIL'),
            "subject": "Welcome!",
            "html": f"<h1>Welcome, {user_name}!</h1>",
            "text": f"Welcome, {user_name}!"
        }
        
        result, error = client.emails.send(payload)
        
        if error:
            raise Exception(f"Failed to send email: {error}")
        
        return result
    except Exception as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

@celery.task(name='tasks.send_bulk_emails')
def send_bulk_emails(email_list: list):
    """Send emails in bulk."""
    batch_items = [
        {
            "to": email['to'],
            "from": os.getenv('UNSENT_FROM_EMAIL'),
            "subject": email['subject'],
            "html": email['html']
        }
        for email in email_list
    ]
    
    result, error = client.emails.batch(batch_items)
    return {"success": not error, "result": result, "error": error}

Use in Routes

routes/auth.py

from flask import Blueprint, request, jsonify
from tasks.email_tasks import send_welcome_email

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    
    # Create user...
    
    # Send email asynchronously
    send_welcome_email.delay(data['email'], data['name'])
    
    return jsonify({"message": "User registered, welcome email queued"}), 201

Run Celery Worker

celery -A celery_app worker --loglevel=info

Webhooks

Handle Unsent webhook events:

routes/webhooks.py

from flask import Blueprint, request, jsonify
import os

webhooks_bp = Blueprint('webhooks', __name__, url_prefix='/webhooks')

@webhooks_bp.route('/unsent', methods=['POST'])
def handle_unsent_webhook():
    """Handle incoming webhook events from Unsent."""
    event = request.json
    event_type = event.get('type')
    data = event.get('data', {})
    
    if event_type == 'email.delivered':
        handle_email_delivered(data)
    elif event_type == 'email.bounced':
        handle_email_bounced(data)
    elif event_type == 'email.opened':
        handle_email_opened(data)
    elif event_type == 'email.clicked':
        handle_email_clicked(data)
    elif event_type == 'email.complained':
        handle_email_complained(data)
    else:
        current_app.logger.info(f"Unhandled event type: {event_type}")
    
    return jsonify({"status": "received"}), 200

def handle_email_delivered(data):
    email_id = data.get('emailId')
    recipient = data.get('recipient')
    current_app.logger.info(f"Email {email_id} delivered to {recipient}")
    # Update database, track analytics, etc.

def handle_email_bounced(data):
    email_id = data.get('emailId')
    recipient = data.get('recipient')
    bounce_type = data.get('bounceType')
    current_app.logger.warning(f"Email {email_id} bounced: {bounce_type}")
    # Mark email as bounced in database

def handle_email_opened(data):
    email_id = data.get('emailId')
    current_app.logger.info(f"Email {email_id} opened")
    # Track open rate

def handle_email_clicked(data):
    email_id = data.get('emailId')
    link = data.get('link')
    current_app.logger.info(f"Link clicked in email {email_id}: {link}")
    # Track click rate

def handle_email_complained(data):
    email_id = data.get('emailId')
    recipient = data.get('recipient')
    current_app.logger.warning(f"Spam complaint for email {email_id}")
    # Unsubscribe user

app.py

from routes.webhooks import webhooks_bp

app.register_blueprint(webhooks_bp)

Testing with Pytest

Write tests for your email functionality:

tests/test_email_service.py

import pytest
from unittest.mock import Mock, patch
from services.email_service import EmailService

@pytest.fixture
def email_service():
    return EmailService()

@pytest.fixture
def mock_unsent_client():
    with patch('services.email_service.unsent') as mock:
        yield mock

def test_send_welcome_email_success(email_service, mock_unsent_client):
    """Test successful welcome email send."""
    # Mock the client response
    mock_client = Mock()
    mock_client.emails.send.return_value = ({"id": "email_123"}, None)
    mock_unsent_client.return_value = mock_client
    
    # Reinitialize service with mocked client
    email_service.client = mock_client
    
    # Send email
    result, error = email_service.send_welcome_email("test@example.com", "Test User")
    
    # Assertions
    assert error is None
    assert result['id'] == "email_123"
    mock_client.emails.send.assert_called_once()

def test_send_welcome_email_failure(email_service, mock_unsent_client):
    """Test failed welcome email send."""
    # Mock the client to return an error
    mock_client = Mock()
    mock_client.emails.send.return_value = (None, {"message": "API Error"})
    email_service.client = mock_client
    
    # Send email
    result, error = email_service.send_welcome_email("test@example.com", "Test User")
    
    # Assertions
    assert result is None
    assert error['message'] == "API Error"

def test_send_password_reset(email_service):
    """Test password reset email."""
    with patch.object(email_service.client.emails, 'send') as mock_send:
        mock_send.return_value = ({"id": "email_456"}, None)
        
        result, error = email_service.send_password_reset("test@example.com", "token123")
        
        assert error is None
        assert result['id'] == "email_456"
        
        # Verify the email contains the reset token
        call_args = mock_send.call_args[0][0]
        assert "token123" in call_args['html']

Run tests:

pytest tests/

Best Practices

1. Configuration Management

Use Flask's config system:

config.py

import os

class Config:
    UNSENT_API_KEY = os.getenv('UNSENT_API_KEY')
    UNSENT_FROM_EMAIL = os.getenv('UNSENT_FROM_EMAIL')
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')

class DevelopmentConfig(Config):
    DEBUG = True
    UNSENT_BASE_URL = 'http://localhost:3004'  # Local API for testing

class ProductionConfig(Config):
    DEBUG = False

config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

Use in app:

app.py

from config import config

app = Flask(__name__)
app.config.from_object(config[os.getenv('FLASK_ENV', 'default')])

2. Error Handling

Create error handlers:

app.py

from flask import jsonify

@app.errorhandler(500)
def internal_error(error):
    app.logger.error(f"Internal error: {error}")
    return jsonify({"error": "Internal server error"}), 500

@app.errorhandler(Exception)
def handle_exception(e):
    app.logger.exception("Unhandled exception")
    return jsonify({"error": str(e)}), 500

3. Logging

Configure structured logging:

app.py

import logging
from logging.handlers import RotatingFileHandler

if not app.debug:
    file_handler = RotatingFileHandler('logs/app.log', maxBytes=1024000, backupCount=10)
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
    ))
    file_handler.setLevel(logging.INFO)
    app.logger.addHandler(file_handler)
    app.logger.setLevel(logging.INFO)
    app.logger.info('Application startup')

Next Steps

Python SDK: Explore all SDK features
FastAPI Guide: Async email sending with FastAPI
Django Guide: Full Django integration

Consider using Flask-Mail compatibility layer if you're migrating from it to Unsent.

Flask

On this page