Flask Production

Complete guide to building scalable production-grade Flask applications

Table of Contents

Flask Architecture

Request/Response Cycle
┌─────────────────────────────────────────────────────────┐
│                     Client Request                       │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                    WSGI Server                           │
│              (Gunicorn, uWSGI, Waitress)                │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                  Flask Application                       │
│  ┌──────────────────────────────────────────────────┐  │
│  │ Application Context                               │  │
│  │ Request Context                                   │  │
│  └──────────────────────────────────────────────────┘  │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│              Before Request Hooks                        │
│  ┌──────────────────────────────────────────────────┐  │
│  │ @app.before_request                               │  │
│  │ @app.before_first_request                         │  │
│  └──────────────────────────────────────────────────┘  │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                    URL Routing                           │
│              (Werkzeug routing system)                   │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                   View Function                          │
│  ┌──────────────────────────────────────────────────┐  │
│  │ • Process request                                 │  │
│  │ • Query database                                  │  │
│  │ • Business logic                                  │  │
│  │ • Generate response                               │  │
│  └──────────────────────────────────────────────────┘  │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│              After Request Hooks                         │
│  ┌──────────────────────────────────────────────────┐  │
│  │ @app.after_request                                │  │
│  │ @app.teardown_request                             │  │
│  └──────────────────────────────────────────────────┘  │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│              Template Rendering                          │
│         (Jinja2, if rendering templates)                │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                    Client Response                       │
└─────────────────────────────────────────────────────────┘
Project Structure (Production)
myproject/
├── app/
│   ├── __init__.py                # Application factory
│   ├── config.py                  # Configuration
│   │
│   ├── api/                       # API blueprints
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── users.py
│   │   └── items.py
│   │
│   ├── main/                      # Main blueprint
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── errors.py
│   │
│   ├── models/                    # Database models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   │
│   ├── schemas/                   # Marshmallow schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   │
│   ├── services/                  # Business logic
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── email_service.py
│   │
│   ├── utils/                     # Utilities
│   │   ├── __init__.py
│   │   ├── decorators.py
│   │   └── validators.py
│   │
│   ├── static/                    # Static files
│   │   ├── css/
│   │   ├── js/
│   │   └── images/
│   │
│   ├── templates/                 # Jinja2 templates
│   │   ├── base.html
│   │   └── index.html
│   │
│   └── extensions.py              # Flask extensions
│
├── migrations/                    # Database migrations
│   └── versions/
│
├── tests/                         # Tests
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_api/
│   ├── test_models/
│   └── test_services/
│
├── instance/                      # Instance folder (not in VCS)
│   └── config.py
│
├── requirements/
│   ├── base.txt
│   ├── development.txt
│   ├── production.txt
│   └── testing.txt
│
├── docker/
│   ├── Dockerfile
│   ├── docker-compose.yml
│   └── nginx.conf
│
├── scripts/
│   ├── init_db.py
│   └── seed_data.py
│
├── .env.example
├── .flaskenv
├── wsgi.py                        # WSGI entry point
├── manage.py                      # CLI commands
├── pytest.ini
└── README.md

Environment Variables (.env)
FLASK_APP=wsgi.py
FLASK_ENV=production
SECRET_KEY=your-secret-key-here

DATABASE_URL=postgresql://user:pass@localhost:5432/db
REDIS_URL=redis://localhost:6379/0

MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
[email protected]
MAIL_PASSWORD=your-password

JWT_SECRET_KEY=your-jwt-secret
Basic Flask Application
# Simple Flask app (not for production)
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/')
def index():
    return jsonify({"message": "Hello, World!"})

@app.route('/users/')
def get_user(user_id):
    return jsonify({"user_id": user_id})

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

# Flask with configuration
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'

# Request handling
from flask import request, jsonify

@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
    if request.method == 'POST':
        data = request.get_json()
        return jsonify({"received": data}), 201
    else:
        return jsonify({"data": "sample"})

# Query parameters
@app.route('/search')
def search():
    query = request.args.get('q', '')
    page = request.args.get('page', 1, type=int)
    return jsonify({"query": query, "page": page})

# Path variables
@app.route('/posts/')
def get_post(post_id):
    return jsonify({"post_id": post_id})

@app.route('/users/')
def get_user_by_name(username):
    return jsonify({"username": username})

# JSON response
@app.route('/api/users')
def get_users():
    users = [
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"}
    ]
    return jsonify(users)

# Response with status code
@app.route('/api/create', methods=['POST'])
def create():
    data = request.get_json()
    # Create resource
    return jsonify({"id": 123}), 201

# Error responses
from flask import abort

@app.route('/api/items/')
def get_item(item_id):
    item = find_item(item_id)
    if not item:
        abort(404)
    return jsonify(item)

# Redirects
from flask import redirect, url_for

@app.route('/old-path')
def old_path():
    return redirect(url_for('new_path'))

@app.route('/new-path')
def new_path():
    return jsonify({"message": "New path"})

# Cookies
from flask import make_response

@app.route('/set-cookie')
def set_cookie():
    response = make_response(jsonify({"message": "Cookie set"}))
    response.set_cookie('user_id', '123', max_age=3600)
    return response

@app.route('/get-cookie')
def get_cookie():
    user_id = request.cookies.get('user_id')
    return jsonify({"user_id": user_id})

# Sessions
from flask import session

app.config['SECRET_KEY'] = 'your-secret-key'

@app.route('/login', methods=['POST'])
def login():
    session['user_id'] = 123
    return jsonify({"message": "Logged in"})

@app.route('/logout')
def logout():
    session.pop('user_id', None)
    return jsonify({"message": "Logged out"})

@app.route('/profile')
def profile():
    user_id = session.get('user_id')
    if not user_id:
        abort(401)
    return jsonify({"user_id": user_id})

Application Factory Pattern

Application Factory
# app/__init__.py
from flask import Flask
from app.extensions import db, migrate, login_manager, cors, ma
from app.config import config

def create_app(config_name='default'):
    """Application factory pattern"""
    app = Flask(__name__)

    # Load configuration
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    # Initialize extensions
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    cors.init_app(app)
    ma.init_app(app)

    # Register blueprints
    from app.api import api_bp
    app.register_blueprint(api_bp, url_prefix='/api')

    from app.main import main_bp
    app.register_blueprint(main_bp)

    # Register error handlers
    register_error_handlers(app)

    # Register CLI commands
    register_cli_commands(app)

    # Setup logging
    setup_logging(app)

    return app

def register_error_handlers(app):
    """Register error handlers"""
    from flask import jsonify

    @app.errorhandler(404)
    def not_found(error):
        return jsonify({"error": "Not found"}), 404

    @app.errorhandler(500)
    def internal_error(error):
        db.session.rollback()
        return jsonify({"error": "Internal server error"}), 500

def register_cli_commands(app):
    """Register custom CLI commands"""
    @app.cli.command()
    def init_db():
        """Initialize the database"""
        db.create_all()
        print("Database initialized")

    @app.cli.command()
    def seed_db():
        """Seed the database with sample data"""
        # Add seed data
        print("Database seeded")

def setup_logging(app):
    """Setup logging"""
    if not app.debug and not app.testing:
        import logging
        from logging.handlers import RotatingFileHandler
        import os

        if not os.path.exists('logs'):
            os.mkdir('logs')

        file_handler = RotatingFileHandler(
            'logs/app.log',
            maxBytes=10240000,
            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')

# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_cors import CORS
from flask_marshmallow import Marshmallow
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
cors = CORS()
ma = Marshmallow()
cache = Cache()
limiter = Limiter(key_func=get_remote_address)

# wsgi.py (entry point)
import os
from app import create_app

app = create_app(os.getenv('FLASK_ENV', 'production'))

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

# Usage in testing
def test_app():
    app = create_app('testing')
    with app.app_context():
        db.create_all()
        # Run tests
        db.drop_all()
Configuration Classes
# app/config.py
import os
from datetime import timedelta

class Config:
    """Base configuration"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_RECORD_QUERIES = True

    # Session
    PERMANENT_SESSION_LIFETIME = timedelta(hours=24)

    # Mail
    MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
    MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() == 'true'
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')

    # Pagination
    ITEMS_PER_PAGE = 20

    # File uploads
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16MB
    UPLOAD_FOLDER = os.path.join(os.getcwd(), 'uploads')

    # Cache
    CACHE_TYPE = 'redis'
    CACHE_REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
    CACHE_DEFAULT_TIMEOUT = 300

    # Rate limiting
    RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/1')

    # JWT
    JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY')
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)

    @staticmethod
    def init_app(app):
        pass

class DevelopmentConfig(Config):
    """Development configuration"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///dev.db'
    SQLALCHEMY_ECHO = True

class TestingConfig(Config):
    """Testing configuration"""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
    WTF_CSRF_ENABLED = False

class ProductionConfig(Config):
    """Production configuration"""
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

    # Security
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Lax'
    REMEMBER_COOKIE_SECURE = True
    REMEMBER_COOKIE_HTTPONLY = True

    @classmethod
    def init_app(cls, app):
        Config.init_app(app)

        # Log to syslog
        import logging
        from logging.handlers import SysLogHandler
        syslog_handler = SysLogHandler()
        syslog_handler.setLevel(logging.WARNING)
        app.logger.addHandler(syslog_handler)

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

# Loading config in app
app = Flask(__name__)
app.config.from_object(config['production'])

# Or from environment variable
config_name = os.getenv('FLASK_ENV', 'production')
app.config.from_object(config[config_name])

# Load from instance folder
app.config.from_pyfile('config.py', silent=True)

# Load from environment variables
app.config.from_prefixed_env()

Blueprints & Modular Design

Blueprint Basics
# app/api/__init__.py
from flask import Blueprint

api_bp = Blueprint('api', __name__)

from app.api import users, items, auth

# app/api/users.py
from flask import jsonify, request
from app.api import api_bp
from app.models import User
from app.extensions import db

@api_bp.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([user.to_dict() for user in users])

@api_bp.route('/users/', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

@api_bp.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    user = User(
        email=data['email'],
        name=data['name']
    )
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201

@api_bp.route('/users/', methods=['PUT'])
def update_user(user_id):
    user = User.query.get_or_404(user_id)
    data = request.get_json()

    user.name = data.get('name', user.name)
    user.email = data.get('email', user.email)

    db.session.commit()
    return jsonify(user.to_dict())

@api_bp.route('/users/', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return '', 204

# Register blueprint in app
from app.api import api_bp
app.register_blueprint(api_bp, url_prefix='/api')

# Multiple blueprints
# Admin blueprint
from flask import Blueprint

admin_bp = Blueprint('admin', __name__)

@admin_bp.route('/dashboard')
def dashboard():
    return jsonify({"page": "admin dashboard"})

# Register with different prefix
app.register_blueprint(admin_bp, url_prefix='/admin')

# Blueprint with subdomain
api_bp = Blueprint('api', __name__, subdomain='api')
app.register_blueprint(api_bp)

# URL: api.example.com/users

# Blueprint with template folder
main_bp = Blueprint(
    'main',
    __name__,
    template_folder='templates',
    static_folder='static'
)

# Blueprint error handlers
@api_bp.errorhandler(404)
def api_not_found(error):
    return jsonify({"error": "API endpoint not found"}), 404

@api_bp.errorhandler(ValidationError)
def validation_error(error):
    return jsonify({"error": str(error)}), 400

# Blueprint before/after request
@api_bp.before_request
def before_api_request():
    # Verify API key
    api_key = request.headers.get('X-API-Key')
    if not verify_api_key(api_key):
        abort(401)

@api_bp.after_request
def after_api_request(response):
    response.headers['X-API-Version'] = '1.0'
    return response
Advanced Blueprint Patterns
# Nested blueprints (Flask 2.0+)
# app/api/v1/__init__.py
from flask import Blueprint

api_v1_bp = Blueprint('api_v1', __name__)

from app.api.v1 import users, items

# app/api/v2/__init__.py
api_v2_bp = Blueprint('api_v2', __name__)

from app.api.v2 import users, items

# Register with different prefixes
app.register_blueprint(api_v1_bp, url_prefix='/api/v1')
app.register_blueprint(api_v2_bp, url_prefix='/api/v2')

# Blueprint factory
def create_api_blueprint(name, version):
    bp = Blueprint(f'api_{name}', __name__)

    @bp.route('/info')
    def info():
        return jsonify({
            "name": name,
            "version": version
        })

    return bp

# Usage
users_v1 = create_api_blueprint('users', 'v1')
users_v2 = create_api_blueprint('users', 'v2')

# Resource-based blueprint pattern
class UserAPI:
    """RESTful user resource"""

    def __init__(self, blueprint):
        self.bp = blueprint
        self.register_routes()

    def register_routes(self):
        self.bp.add_url_rule(
            '/users',
            'list_users',
            self.list_users,
            methods=['GET']
        )
        self.bp.add_url_rule(
            '/users',
            'create_user',
            self.create_user,
            methods=['POST']
        )
        self.bp.add_url_rule(
            '/users/',
            'get_user',
            self.get_user,
            methods=['GET']
        )

    def list_users(self):
        users = User.query.all()
        return jsonify([u.to_dict() for u in users])

    def create_user(self):
        data = request.get_json()
        user = User(**data)
        db.session.add(user)
        db.session.commit()
        return jsonify(user.to_dict()), 201

    def get_user(self, user_id):
        user = User.query.get_or_404(user_id)
        return jsonify(user.to_dict())

# Usage
api_bp = Blueprint('api', __name__)
UserAPI(api_bp)

# Blueprint with context processor
@main_bp.context_processor
def inject_user():
    return dict(current_user=get_current_user())

# Blueprint with URL value preprocessor
@api_bp.url_value_preprocessor
def get_api_version(endpoint, values):
    values['api_version'] = request.headers.get('API-Version', '1.0')

@api_bp.route('/data')
def get_data(api_version):
    return jsonify({"api_version": api_version})

# Blueprint with template filter
@main_bp.app_template_filter('format_currency')
def format_currency(value):
    return f"${value:,.2f}"

# Modular app structure with blueprints
# app/modules/users/routes.py
from flask import Blueprint

users_bp = Blueprint('users', __name__)

@users_bp.route('/')
def list_users():
    pass

@users_bp.route('/')
def get_user(user_id):
    pass

# app/modules/users/__init__.py
from app.modules.users.routes import users_bp

# Register all module blueprints
from app.modules.users import users_bp
from app.modules.items import items_bp
from app.modules.auth import auth_bp

app.register_blueprint(users_bp, url_prefix='/users')
app.register_blueprint(items_bp, url_prefix='/items')
app.register_blueprint(auth_bp, url_prefix='/auth')

Configuration Management

Configuration Best Practices
# Using python-dotenv
# .env file
DATABASE_URL=postgresql://localhost/mydb
SECRET_KEY=my-secret-key
DEBUG=True

# Load in app
from dotenv import load_dotenv
import os

load_dotenv()

app.config['DATABASE_URL'] = os.getenv('DATABASE_URL')
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY')
app.config['DEBUG'] = os.getenv('DEBUG', 'False') == 'True'

# Environment-specific configs
import os

class Config:
    """Base config"""
    SECRET_KEY = os.environ.get('SECRET_KEY')
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    @staticmethod
    def init_app(app):
        pass

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL')
    SQLALCHEMY_ECHO = True

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

    @classmethod
    def init_app(cls, app):
        Config.init_app(app)

        # Email errors to admins
        import logging
        from logging.handlers import SMTPHandler
        credentials = None
        secure = None

        if getattr(cls, 'MAIL_USERNAME', None):
            credentials = (cls.MAIL_USERNAME, cls.MAIL_PASSWORD)
            if getattr(cls, 'MAIL_USE_TLS', None):
                secure = ()

        mail_handler = SMTPHandler(
            mailhost=(cls.MAIL_SERVER, cls.MAIL_PORT),
            fromaddr=cls.MAIL_SENDER,
            toaddrs=[cls.ADMIN_EMAIL],
            subject='Application Error',
            credentials=credentials,
            secure=secure
        )
        mail_handler.setLevel(logging.ERROR)
        app.logger.addHandler(mail_handler)

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

# Configuration from file
# config.py
DEBUG = False
SECRET_KEY = 'secret'
DATABASE_URI = 'sqlite:///app.db'

# Load in app
app.config.from_pyfile('config.py')

# Configuration from object
class Config:
    DEBUG = False
    TESTING = False
    DATABASE_URI = 'sqlite:///app.db'

app.config.from_object(Config)

# Configuration from JSON
# config.json
{
    "DEBUG": false,
    "SECRET_KEY": "secret",
    "DATABASE_URI": "sqlite:///app.db"
}

# Load
import json
with open('config.json') as f:
    config = json.load(f)
app.config.update(config)

# Instance folder configuration
# instance/config.py (not in version control)
SECRET_KEY = 'production-secret-key'
DATABASE_URI = 'postgresql://production-db'

# Load
app.config.from_pyfile('config.py', silent=True)

# Accessing configuration
from flask import current_app

def some_function():
    secret_key = current_app.config['SECRET_KEY']
    debug_mode = current_app.config.get('DEBUG', False)

# Custom configuration loader
class ConfigLoader:
    @staticmethod
    def load_config(app, config_name):
        # Load base config
        app.config.from_object(config[config_name])

        # Load instance config
        app.config.from_pyfile('config.py', silent=True)

        # Load environment variables
        app.config.update({
            key[5:]: value
            for key, value in os.environ.items()
            if key.startswith('FLASK_')
        })

        # Validate required config
        required = ['SECRET_KEY', 'DATABASE_URL']
        for key in required:
            if not app.config.get(key):
                raise ValueError(f"Missing required config: {key}")

Database Integration

SQLAlchemy Models
# app/models/user.py
from app.extensions import db
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))
    name = db.Column(db.String(100))
    is_active = db.Column(db.Boolean, default=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # Relationships
    posts = db.relationship('Post', backref='author', lazy='dynamic')
    profile = db.relationship('Profile', backref='user', uselist=False)

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

    def to_dict(self):
        return {
            'id': self.id,
            'email': self.email,
            'username': self.username,
            'name': self.name,
            'created_at': self.created_at.isoformat()
        }

    def __repr__(self):
        return f''

# One-to-Many relationship
class Post(db.Model):
    __tablename__ = 'posts'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    # Cascade delete
    comments = db.relationship(
        'Comment',
        backref='post',
        lazy='dynamic',
        cascade='all, delete-orphan'
    )

# Many-to-Many relationship
# Association table
tags = db.Table('post_tags',
    db.Column('post_id', db.Integer, db.ForeignKey('posts.id'), primary_key=True),
    db.Column('tag_id', db.Integer, db.ForeignKey('tags.id'), primary_key=True)
)

class Post(db.Model):
    # ... other fields
    tags = db.relationship(
        'Tag',
        secondary=tags,
        lazy='subquery',
        backref=db.backref('posts', lazy=True)
    )

class Tag(db.Model):
    __tablename__ = 'tags'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False)

# Query patterns
# Basic queries
user = User.query.get(1)  # By primary key
user = User.query.get_or_404(1)  # Or 404 error
users = User.query.all()  # All records
user = User.query.first()  # First record
count = User.query.count()  # Count

# Filtering
users = User.query.filter_by(is_active=True).all()
users = User.query.filter(User.is_active == True).all()
users = User.query.filter(User.name.like('%John%')).all()

# Ordering
users = User.query.order_by(User.created_at.desc()).all()
users = User.query.order_by(User.name, User.created_at).all()

# Pagination
page = 1
per_page = 20
users = User.query.paginate(page=page, per_page=per_page)

for user in users.items:
    print(user.username)

# Pagination info
print(f"Page {users.page} of {users.pages}")
print(f"Total: {users.total}")
print(f"Has next: {users.has_next}")
print(f"Has prev: {users.has_prev}")

# Joins
posts = Post.query.join(User).filter(User.is_active == True).all()

# Aggregation
from sqlalchemy import func

user_count_by_date = db.session.query(
    func.date(User.created_at).label('date'),
    func.count(User.id).label('count')
).group_by('date').all()

# CRUD operations
# Create
user = User(email='[email protected]', username='user')
user.set_password('password')
db.session.add(user)
db.session.commit()

# Read
user = User.query.filter_by(email='[email protected]').first()

# Update
user.name = 'New Name'
db.session.commit()

# Delete
db.session.delete(user)
db.session.commit()

# Bulk insert
users = [
    User(email=f'user{i}@example.com', username=f'user{i}')
    for i in range(100)
]
db.session.bulk_save_objects(users)
db.session.commit()

# Transaction management
try:
    user = User(email='[email protected]')
    db.session.add(user)

    profile = Profile(user=user, bio='Test bio')
    db.session.add(profile)

    db.session.commit()
except Exception as e:
    db.session.rollback()
    raise

# Raw SQL
from sqlalchemy import text

result = db.session.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": "[email protected]"}
)
user = result.fetchone()
Database Migrations
# Flask-Migrate setup
from flask_migrate import Migrate

migrate = Migrate(app, db)

# Initialize migrations
flask db init

# Create migration
flask db migrate -m "Initial migration"

# Apply migration
flask db upgrade

# Rollback migration
flask db downgrade

# Show migration history
flask db history

# Show current revision
flask db current

# Manual migration file
# migrations/versions/xxx_add_user_role.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column(
        'users',
        sa.Column('role', sa.String(20), nullable=True)
    )
    op.create_index('ix_users_role', 'users', ['role'])

def downgrade():
    op.drop_index('ix_users_role', 'users')
    op.drop_column('users', 'role')

# Data migration
def upgrade():
    # Add column
    op.add_column('users', sa.Column('email_verified', sa.Boolean))

    # Set default values
    op.execute("UPDATE users SET email_verified = FALSE")

    # Make non-nullable
    op.alter_column('users', 'email_verified', nullable=False)

# Migration with SQLAlchemy
from sqlalchemy.orm import Session

def upgrade():
    bind = op.get_bind()
    session = Session(bind=bind)

    # Complex data transformation
    users = session.execute("SELECT id, old_field FROM users")
    for user in users:
        new_value = transform(user.old_field)
        session.execute(
            "UPDATE users SET new_field = :val WHERE id = :id",
            {"val": new_value, "id": user.id}
        )

    session.commit()

Authentication & Authorization

Flask-Login
# Setup Flask-Login
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

# User model with Flask-Login
class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True)
    password_hash = db.Column(db.String(128))

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

# Login route
from flask_login import login_user, current_user

@auth_bp.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(email=data['email']).first()

    if user is None or not user.check_password(data['password']):
        return jsonify({"error": "Invalid credentials"}), 401

    login_user(user, remember=data.get('remember', False))
    return jsonify({"message": "Logged in successfully"})

# Logout route
from flask_login import logout_user

@auth_bp.route('/logout', methods=['POST'])
@login_required
def logout():
    logout_user()
    return jsonify({"message": "Logged out successfully"})

# Protected route
from flask_login import login_required, current_user

@app.route('/profile')
@login_required
def profile():
    return jsonify(current_user.to_dict())

# JWT Authentication
from flask_jwt_extended import (
    JWTManager,
    create_access_token,
    create_refresh_token,
    jwt_required,
    get_jwt_identity
)

jwt = JWTManager(app)

@auth_bp.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(email=data['email']).first()

    if not user or not user.check_password(data['password']):
        return jsonify({"error": "Invalid credentials"}), 401

    access_token = create_access_token(identity=user.id)
    refresh_token = create_refresh_token(identity=user.id)

    return jsonify({
        "access_token": access_token,
        "refresh_token": refresh_token
    })

@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    user_id = get_jwt_identity()
    access_token = create_access_token(identity=user_id)
    return jsonify({"access_token": access_token})

@app.route('/protected')
@jwt_required()
def protected():
    user_id = get_jwt_identity()
    user = User.query.get(user_id)
    return jsonify(user.to_dict())

# Custom decorators
from functools import wraps
from flask import abort

def admin_required(f):
    @wraps(f)
    @login_required
    def decorated_function(*args, **kwargs):
        if not current_user.is_admin:
            abort(403)
        return f(*args, **kwargs)
    return decorated_function

@app.route('/admin/dashboard')
@admin_required
def admin_dashboard():
    return jsonify({"message": "Admin dashboard"})

# Permission decorator
def permission_required(permission):
    def decorator(f):
        @wraps(f)
        @login_required
        def decorated_function(*args, **kwargs):
            if not current_user.has_permission(permission):
                abort(403)
            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route('/posts', methods=['POST'])
@permission_required('create_post')
def create_post():
    pass

Resources & Learning Path

Learning Progression
Phase 1: Flask Fundamentals (1-2 weeks)
□ Basic routing and views
□ Templates with Jinja2
□ Forms and validation
□ Request/response handling
□ Static files

Phase 2: Intermediate Flask (2-3 weeks)
□ Blueprints
□ Database with SQLAlchemy
□ Authentication
□ RESTful APIs
□ Error handling

Phase 3: Advanced Flask (3-4 weeks)
□ Application factory pattern
□ Flask extensions
□ Testing strategies
□ Background tasks
□ WebSockets

Phase 4: Production Flask (Ongoing)
□ Deployment strategies
□ Performance optimization
□ Security hardening
□ Monitoring and logging
□ Microservices patterns
Related Comprehensive Sheets
Python Ecosystem
→ Python Advanced (decorators, context managers)
→ Django Production (alternative framework)
→ FastAPI Advanced (modern async framework)
→ SQLAlchemy Advanced
→ Celery & Task Queues

Web Development
→ REST API Design
→ WebSockets
→ GraphQL
→ Authentication Patterns
→ API Security

Testing & Quality
→ Pytest Advanced
→ Integration Testing
→ Load Testing
→ API Testing

Deployment
→ Docker & Kubernetes
→ CI/CD Pipelines
→ Nginx Configuration
→ Monitoring & Observability
Pro Tips Summary
Architecture
✓ Use application factory pattern
✓ Organize code with blueprints
✓ Separate configuration by environment
✓ Keep business logic in services
✓ Use extensions for common functionality

Database
✓ Use migrations for schema changes
✓ Index frequently queried columns
✓ Use eager loading to avoid N+1
✓ Implement proper transaction handling
✓ Use connection pooling

Security
✓ Never commit secrets to version control
✓ Use environment variables
✓ Implement proper authentication
✓ Validate all user input
✓ Use HTTPS in production
✓ Implement rate limiting

Performance
✓ Use caching strategically
✓ Optimize database queries
✓ Enable compression
✓ Use CDN for static files
✓ Profile and monitor performance

Testing
✓ Write unit tests for all routes
✓ Test error conditions
✓ Use fixtures for test data
✓ Mock external services
✓ Measure code coverage