Flask Integration

Complete Flask 3.0 integration with Application Factory pattern, Blueprints, validation, security headers, rate limiting, and comprehensive testing.

This example uses Flask 3.0+ patterns including Application Factory, Blueprints, and modern extension patterns.

Project Structure

myapp/
├── app/
│   ├── __init__.py              # Application factory
│   ├── extensions.py            # Flask extensions
│   ├── config.py                # Environment configurations
│   ├── logging_config.py        # Logging setup
│   ├── sent_integration.py      # Sent Flask extension
│   ├── blueprints/
│   │   ├── __init__.py
│   │   ├── messages.py          # Message routes
│   │   └── webhooks.py          # Webhook handlers
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── message_schemas.py   # Marshmallow schemas
│   ├── services/
│   │   ├── __init__.py
│   │   └── message_service.py   # Business logic
│   └── utils/
│       ├── __init__.py
│       └── decorators.py        # Custom decorators
├── tests/
│   ├── __init__.py
│   ├── conftest.py              # Pytest fixtures
│   ├── test_messages.py
│   └── test_webhooks.py
├── wsgi.py                      # WSGI entry point
├── gunicorn.conf.py             # Gunicorn configuration
├── requirements.txt
└── .env

Installation

pip install flask sentdm marshmallow flask-marshmallow flask-limiter flask-talisman
pip install pydantic python-dotenv gunicorn pytest pytest-flask

Configuration

# app/config.py
import os
from datetime import timedelta

class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')
    SENT_DM_API_KEY = os.getenv('SENT_DM_API_KEY')
    SENT_DM_WEBHOOK_SECRET = os.getenv('SENT_DM_WEBHOOK_SECRET')
    RATELIMIT_STORAGE_URI = os.getenv('RATELIMIT_STORAGE_URI', 'memory://')
    RATELIMIT_STRATEGY = 'fixed-window'
    RATELIMIT_DEFAULT = '100/minute'
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Lax'
    PERMANENT_SESSION_LIFETIME = timedelta(hours=24)
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')

class DevelopmentConfig(Config):
    DEBUG = True
    SESSION_COOKIE_SECURE = False
    LOG_LEVEL = 'DEBUG'

class TestingConfig(Config):
    TESTING = True
    DEBUG = True
    SENT_DM_API_KEY = 'test-api-key'
    SENT_DM_WEBHOOK_SECRET = 'test-webhook-secret'
    RATELIMIT_ENABLED = False
    SESSION_COOKIE_SECURE = False

class ProductionConfig(Config):
    DEBUG = False
    LOG_LEVEL = 'WARNING'

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

Logging Configuration

# app/logging_config.py
import logging
import sys
from logging.handlers import RotatingFileHandler

def configure_logging(app):
    log_level = getattr(logging, app.config['LOG_LEVEL'], logging.INFO)
    app.logger.setLevel(log_level)
    app.logger.handlers.clear()

    console = logging.StreamHandler(sys.stdout)
    console.setFormatter(logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    ))
    app.logger.addHandler(console)

    if not app.debug and not app.testing:
        file_handler = RotatingFileHandler('app.log', maxBytes=10485760, backupCount=10)
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

Flask Extension for Sent

# app/sent_integration.py
from flask import current_app, g
from sent_dm import SentDm

class SentExtension:
    def __init__(self, app=None):
        self.app = app
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        app.extensions = getattr(app, 'extensions', {})
        app.extensions['sent'] = self
        app.teardown_appcontext(self.teardown)

    def teardown(self, exception):
        pass

    @property
    def client(self):
        if '_sent_client' not in g:
            api_key = current_app.config.get('SENT_DM_API_KEY')
            if not api_key:
                raise RuntimeError("SENT_DM_API_KEY not configured")
            g._sent_client = SentDm(api_key)
        return g._sent_client

sent_extension = SentExtension()

def get_sent_client():
    return current_app.extensions['sent'].client

Extensions Setup

# app/extensions.py
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_talisman import Talisman
from flask_marshmallow import Marshmallow

limiter = Limiter(key_func=get_remote_address, default_limits=["100 per minute"])
talisman = Talisman()
ma = Marshmallow()

Marshmallow Schemas

# app/schemas/message_schemas.py
from marshmallow import Schema, fields, validate

class SendMessageSchema(Schema):
    phone_number = fields.String(
        required=True,
        validate=validate.Regexp(r'^\+[1-9]\d{1,14}$', error="E.164 format required")
    )
    template_id = fields.String(required=True, validate=validate.Length(min=1, max=100))
    template_name = fields.String(required=True, validate=validate.Length(min=1, max=100))
    parameters = fields.Dict(keys=fields.String(), values=fields.String(), load_default=dict)
    channels = fields.List(
        fields.String(validate=validate.OneOf(['sms', 'whatsapp', 'email'])),
        load_default=list
    )

class WelcomeMessageSchema(Schema):
    phone_number = fields.String(required=True, validate=validate.Regexp(r'^\+[1-9]\d{1,14}$'))
    name = fields.String(load_default="Valued Customer", validate=validate.Length(max=100))

class OrderConfirmationSchema(Schema):
    phone_number = fields.String(required=True, validate=validate.Regexp(r'^\+[1-9]\d{1,14}$'))
    order_number = fields.String(required=True)
    total = fields.String(required=True)

class MessageResponseSchema(Schema):
    message_id = fields.String(dump_only=True)
    status = fields.String(dump_only=True)
    sent_at = fields.DateTime(dump_only=True)

class WebhookEventSchema(Schema):
    type = fields.String(required=True)
    data = fields.Dict(required=True)
    timestamp = fields.DateTime(dump_only=True)

send_message_schema = SendMessageSchema()
welcome_message_schema = WelcomeMessageSchema()
order_confirmation_schema = OrderConfirmationSchema()
message_response_schema = MessageResponseSchema()
webhook_event_schema = WebhookEventSchema()

Business Logic Service

# app/services/message_service.py
import logging
from flask import current_app
from app.sent_integration import get_sent_client

logger = logging.getLogger(__name__)

class MessageService:
    def __init__(self):
        self.sent = None

    def _get_client(self):
        if self.sent is None:
            self.sent = get_sent_client()
        return self.sent

    def send_message(self, phone_number: str, template_id: str,
                     template_name: str, parameters: dict = None,
                     channels: list = None) -> dict:
        try:
            client = self._get_client()
            response = client.messages.send(
                to=[phone_number],
                template={'id': template_id, 'name': template_name, 'parameters': parameters or {}},
                channels=channels
            )
            message = response.data.messages[0]
            logger.info(f"Message sent: {message.id} to {phone_number}")
            return {'message_id': message.id, 'status': message.status, 'success': True}
        except Exception as e:
            logger.error(f"Failed to send message: {str(e)}")
            raise

    def send_welcome_message(self, phone_number: str, name: str = "Valued Customer") -> dict:
        return self.send_message(
            phone_number=phone_number, template_id='welcome-template-id',
            template_name='welcome', parameters={'name': name}, channels=['whatsapp']
        )

    def send_order_confirmation(self, phone_number: str, order_number: str, total: str) -> dict:
        return self.send_message(
            phone_number=phone_number, template_id='order-confirmation-id',
            template_name='order_confirmation',
            parameters={'order_number': order_number, 'total': total},
            channels=['sms', 'whatsapp']
        )

message_service = MessageService()

Custom Decorators

# app/utils/decorators.py
import functools
import hmac
import hashlib
from flask import request, jsonify, current_app

def verify_webhook_signature(f):
    @functools.wraps(f)
    def decorated_function(*args, **kwargs):
        signature = request.headers.get('X-Webhook-Signature')
        if not signature:
            return jsonify({'error': 'Missing webhook signature'}), 401
        webhook_secret = current_app.config.get('SENT_DM_WEBHOOK_SECRET')
        if not webhook_secret:
            return jsonify({'error': 'Webhook secret not configured'}), 500
        expected = hmac.new(
            webhook_secret.encode('utf-8'), request.get_data(), hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(signature, expected):
            return jsonify({'error': 'Invalid signature'}), 401
        return f(*args, **kwargs)
    return decorated_function

def handle_validation_errors(f):
    @functools.wraps(f)
    def decorated_function(*args, **kwargs):
        from marshmallow import ValidationError
        try:
            return f(*args, **kwargs)
        except ValidationError as err:
            return jsonify({'error': 'Validation error', 'messages': err.messages}), 400
    return decorated_function

Blueprints

Messages Blueprint

# app/blueprints/messages.py
from flask import Blueprint, request, jsonify, current_app
from marshmallow import ValidationError
from app.extensions import limiter, ma
from app.schemas.message_schemas import (
    send_message_schema, welcome_message_schema,
    order_confirmation_schema, message_response_schema
)
from app.services.message_service import message_service
from app.utils.decorators import handle_validation_errors

messages_bp = Blueprint('messages', __name__, url_prefix='/api/messages')

@messages_bp.errorhandler(ValidationError)
def handle_validation_error(error):
    return jsonify({'error': 'Validation error', 'messages': error.messages}), 400

@messages_bp.errorhandler(Exception)
def handle_generic_error(error):
    current_app.logger.error(f"Error: {str(error)}", exc_info=True)
    return jsonify({'error': 'Internal server error'}), 500

@messages_bp.route('/send', methods=['POST'])
@limiter.limit("10 per minute")
@handle_validation_errors
def send_message():
    json_data = request.get_json()
    if not json_data:
        return jsonify({'error': 'No input data provided'}), 400
    data = send_message_schema.load(json_data)
    result = message_service.send_message(
        phone_number=data['phone_number'], template_id=data['template_id'],
        template_name=data['template_name'], parameters=data.get('parameters'),
        channels=data.get('channels')
    )
    return jsonify(message_response_schema.dump(result)), 200

@messages_bp.route('/welcome', methods=['POST'])
@limiter.limit("10 per minute")
@handle_validation_errors
def send_welcome():
    data = welcome_message_schema.load(request.get_json() or {})
    result = message_service.send_welcome_message(
        phone_number=data['phone_number'], name=data.get('name', 'Valued Customer')
    )
    return jsonify(message_response_schema.dump(result)), 200

@messages_bp.route('/order-confirmation', methods=['POST'])
@limiter.limit("10 per minute")
@handle_validation_errors
def send_order_confirmation():
    data = order_confirmation_schema.load(request.get_json() or {})
    result = message_service.send_order_confirmation(
        phone_number=data['phone_number'],
        order_number=data['order_number'], total=data['total']
    )
    return jsonify(message_response_schema.dump(result)), 200

Webhooks Blueprint

# app/blueprints/webhooks.py
import logging
from flask import Blueprint, request, jsonify, current_app
from app.extensions import limiter
from app.utils.decorators import verify_webhook_signature
from app.schemas.message_schemas import webhook_event_schema

logger = logging.getLogger(__name__)
webhooks_bp = Blueprint('webhooks', __name__, url_prefix='/webhooks')

@webhooks_bp.route('/sent', methods=['POST'])
@limiter.limit("100 per minute")
@verify_webhook_signature
def handle_sent_webhook():
    event_data = request.get_json()
    if not event_data:
        return jsonify({'error': 'Invalid JSON payload'}), 400
    try:
        event = webhook_event_schema.load(event_data)
        process_webhook_event(event)
        return jsonify({'received': True}), 200
    except Exception as e:
        logger.error(f"Error processing webhook: {str(e)}")
        return jsonify({'received': True}), 200

def process_webhook_event(event: dict):
    event_type = event.get('type')
    data = event.get('data', {})
    logger.info(f"Processing webhook: {event_type}")
    match event_type:
        case 'message.status.updated':
            logger.info(f"Message {data.get('id')} status: {data.get('status')}")
        case 'message.delivered':
            logger.info(f"Message {data.get('id')} delivered")
        case 'message.failed':
            logger.error(f"Message {data.get('id')} failed: {data.get('error', {})}")
        case 'message.read':
            logger.info(f"Message {data.get('id')} read")
        case _:
            logger.warning(f"Unhandled event: {event_type}")

Application Factory

# app/__init__.py
from flask import Flask, jsonify
from app.config import config
from app.extensions import limiter, talisman, ma, sent_extension
from app.logging_config import configure_logging
from app.blueprints.messages import messages_bp
from app.blueprints.webhooks import webhooks_bp

def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    configure_logging(app)

    limiter.init_app(app)
    ma.init_app(app)
    sent_extension.init_app(app)
    talisman.init_app(app, force_https=False, strict_transport_security=True,
        content_security_policy={'default-src': "'self'", 'script-src': "'self'"},
        referrer_policy='strict-origin-when-cross-origin')

    app.register_blueprint(messages_bp)
    app.register_blueprint(webhooks_bp)
    register_error_handlers(app)

    @app.route('/health')
    def health_check():
        return jsonify({'status': 'healthy', 'service': 'sent-flask-integration'})

    @app.route('/')
    def index():
        return jsonify({
            'service': 'Sent DM Flask Integration',
            'endpoints': {'health': '/health', 'messages': '/api/messages', 'webhooks': '/webhooks/sent'}
        })

    return app

def register_error_handlers(app):
    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({'error': 'Bad request'}), 400
    @app.errorhandler(429)
    def rate_limit_handler(error):
        return jsonify({'error': 'Rate limit exceeded', 'retry_after': error.description}), 429
    @app.errorhandler(500)
    def internal_error(error):
        app.logger.error(f"Server error: {str(error)}", exc_info=True)
        return jsonify({'error': 'Internal server error'}), 500

WSGI Entry Point

# wsgi.py
import os
from app import create_app

config_name = os.getenv('FLASK_CONFIG', 'production')
app = create_app(config_name)

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

Gunicorn Configuration

# gunicorn.conf.py
import multiprocessing
import os

bind = os.getenv('GUNICORN_BIND', '0.0.0.0:8000')
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = 'sync'
worker_connections = 1000
timeout = 30
keepalive = 2
accesslog = '-'
errorlog = '-'
loglevel = os.getenv('GUNICORN_LOG_LEVEL', 'info')
preload_app = True
keyfile = os.getenv('SSL_KEY_FILE')
certfile = os.getenv('SSL_CERT_FILE')

Testing

Pytest Configuration

# tests/conftest.py
import pytest
from app import create_app
from app.extensions import limiter

@pytest.fixture
def app():
    app = create_app('testing')
    limiter.enabled = False
    with app.app_context():
        yield app

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def auth_headers():
    return {'Content-Type': 'application/json'}

@pytest.fixture
def mock_sent_client(mocker):
    mock = mocker.MagicMock()
    mock.messages.send.return_value = mocker.MagicMock(
        data=mocker.MagicMock(messages=[mocker.MagicMock(id='msg_123', status='pending')])
    )
    return mock

Message Tests

# tests/test_messages.py
import json
import pytest
from unittest.mock import patch

class TestSendMessage:
    def test_send_message_success(self, client, auth_headers, mocker):
        mock_response = mocker.MagicMock()
        mock_response.data.messages = [mocker.MagicMock(id='msg_123', status='pending')]
        with patch('app.services.message_service.get_sent_client') as mock_get_client:
            mock_client = mocker.MagicMock()
            mock_client.messages.send.return_value = mock_response
            mock_get_client.return_value = mock_client
            response = client.post('/api/messages/send', data=json.dumps({
                'phone_number': '+1234567890', 'template_id': 'template-123',
                'template_name': 'welcome', 'parameters': {'name': 'John'}
            }), headers=auth_headers)
            assert response.status_code == 200
            data = json.loads(response.data)
            assert data['message_id'] == 'msg_123'

    def test_send_message_validation_error(self, client, auth_headers):
        response = client.post('/api/messages/send', data=json.dumps({
            'phone_number': 'invalid', 'template_id': 'template-123', 'template_name': 'welcome'
        }), headers=auth_headers)
        assert response.status_code == 400

    def test_send_message_missing_fields(self, client, auth_headers):
        response = client.post('/api/messages/send', data=json.dumps({}), headers=auth_headers)
        assert response.status_code == 400

class TestWelcomeMessage:
    def test_send_welcome_success(self, client, auth_headers, mocker):
        mock_response = mocker.MagicMock()
        mock_response.data.messages = [mocker.MagicMock(id='msg_456', status='queued')]
        with patch('app.services.message_service.get_sent_client') as mock_get_client:
            mock_client = mocker.MagicMock()
            mock_client.messages.send.return_value = mock_response
            mock_get_client.return_value = mock_client
            response = client.post('/api/messages/welcome', data=json.dumps({
                'phone_number': '+1234567890', 'name': 'Jane'
            }), headers=auth_headers)
            assert response.status_code == 200
            call_args = mock_client.messages.send.call_args
            assert call_args[1]['template']['name'] == 'welcome'

Webhook Tests

# tests/test_webhooks.py
import json
import hmac
import hashlib
import pytest
from unittest.mock import patch

class TestWebhookHandler:
    def _generate_signature(self, payload: str, secret: str) -> str:
        return hmac.new(secret.encode('utf-8'), payload.encode('utf-8'), hashlib.sha256).hexdigest()

    def test_webhook_missing_signature(self, client):
        response = client.post('/webhooks/sent', data='{}')
        assert response.status_code == 401

    def test_webhook_invalid_signature(self, client):
        response = client.post('/webhooks/sent', data='{}',
            headers={'X-Webhook-Signature': 'invalid'})
        assert response.status_code == 401

    def test_webhook_valid_signature(self, client, app):
        payload = json.dumps({'type': 'message.delivered', 'data': {'id': 'msg_123'}})
        signature = self._generate_signature(payload, app.config['SENT_DM_WEBHOOK_SECRET'])
        response = client.post('/webhooks/sent', data=payload,
            headers={'X-Webhook-Signature': signature})
        assert response.status_code == 200

    def test_webhook_message_failed(self, client, app):
        payload = json.dumps({'type': 'message.failed', 'data': {'id': 'msg_123', 'error': {'message': 'Invalid'}}})
        signature = self._generate_signature(payload, app.config['SENT_DM_WEBHOOK_SECRET'])
        with patch('app.blueprints.webhooks.logger') as mock_logger:
            response = client.post('/webhooks/sent', data=payload,
                headers={'X-Webhook-Signature': signature})
            assert response.status_code == 200
            mock_logger.error.assert_called_once()

Running the Application

Development

export FLASK_APP=wsgi.py
export FLASK_CONFIG=development
export SENT_DM_API_KEY=your_api_key
export SENT_DM_WEBHOOK_SECRET=your_webhook_secret
flask run

Production

gunicorn -c gunicorn.conf.py wsgi:app

Docker

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["gunicorn", "-c", "gunicorn.conf.py", "wsgi:app"]

Environment Variables

# Required
SENT_DM_API_KEY=your_api_key_here
SENT_DM_WEBHOOK_SECRET=your_webhook_secret

# Optional
FLASK_CONFIG=production
SECRET_KEY=your-secret-key
LOG_LEVEL=INFO
RATELIMIT_STORAGE_URI=redis://localhost:6379
SSL_KEY_FILE=/path/to/key.pem
SSL_CERT_FILE=/path/to/cert.pem

Rate Limiting

Endpoint	Limit
`/api/messages/*`	10 requests/minute
`/webhooks/sent`	100 requests/minute
Other endpoints	100 requests/minute (default)

Configure Redis for distributed rate limiting:

RATELIMIT_STORAGE_URI=redis://redis:6379/0

Security Headers

Flask-Talisman adds: Strict-Transport-Security, Content-Security-Policy, X-Content-Type-Options, X-Frame-Options, X-XSS-Protection, Referrer-Policy

Next Steps

Learn about best practices for production deployments
Set up webhooks for delivery status updates
Explore the Python SDK reference for advanced features
Review Celery integration for background processing

On this page