Post

🌊 Flask: Deep Dive & Best Practices

Concise, clear, and validated revision notes on using Flask framework - application structure, request handling, template rendering, and essential security and deployment best practices.

🌊 Flask: Deep Dive & Best Practices

Table of Contents

  1. Introduction
  2. Core Concepts
  3. Application Structure
  4. Application Factory Pattern
  5. Blueprints
  6. Configuration Management
  7. Database Integration
  8. Request Handling
  9. Templates and Static Files
  10. REST API Development
  11. Authentication and Authorization
  12. Security Best Practices
  13. Testing
  14. Deployment
  15. Performance Optimization
  16. Best Practices
  17. Common Pitfalls
  18. Jargon Tables

Introduction

Flask is a lightweight, flexible, and powerful micro web framework for Python. Created by Armin Ronacher, Flask follows the WSGI (Web Server Gateway Interface) specification and is based on the Werkzeug WSGI toolkit and Jinja2 template engine. Unlike full-stack frameworks like Django, Flask provides only the essentials, allowing developers to choose their own tools and libraries for specific functionalities.

Key Features

  • Micro-framework: Minimal core with extensibility through extensions
  • WSGI Compliance: Standard Python web server interface
  • Built-in Development Server: Rapid prototyping and testing
  • Jinja2 Templating: Powerful template engine with inheritance
  • RESTful Request Dispatching: Clean URL routing
  • Secure Cookies: Session management support
  • Unit Testing Support: Built-in test client
  • Extensive Extensions: Database, authentication, forms, and more
  • Production Ready: Powers thousands of applications worldwide

When to Use Flask

  • RESTful APIs: Microservices and API backends
  • Small to Medium Applications: When you need flexibility
  • Prototyping: Quick proof-of-concept development
  • Learning: Understanding web application fundamentals
  • Custom Requirements: When you need fine-grained control

Core Concepts

WSGI (Web Server Gateway Interface)

WSGI is the standard interface between web servers and Python web applications. Flask applications are WSGI applications that can be deployed to any WSGI-compatible server.

1
2
3
4
5
6
7
8
9
from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello, World!'

# The app object is a WSGI application

Request-Response Cycle

Flask follows a simple request-response pattern:

  1. Client Request: Browser/client sends HTTP request
  2. URL Routing: Flask matches URL to a view function
  3. View Processing: Function processes request and generates response
  4. Response Return: Flask sends HTTP response back to client

Application Context

Flask uses contexts to make certain variables globally accessible during request handling:

Application Context

Contains application-level data and is active during requests or CLI commands.

1
2
3
4
5
from flask import current_app

# Access configuration
with app.app_context():
    print(current_app.config['SECRET_KEY'])

Request Context

Contains request-level data and is active during request processing.

1
2
3
4
5
6
7
from flask import request

@app.route('/user/<username>')
def show_user(username):
    # Access request data
    user_agent = request.headers.get('User-Agent')
    return f'Hello {username}, using {user_agent}'

View Functions

View functions (also called route handlers) are Python functions mapped to URLs that handle requests and return responses.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@app.route('/hello')
def hello():
    """Simple view returning string"""
    return 'Hello!'

@app.route('/json')
def json_response():
    """View returning JSON"""
    return {'message': 'Hello', 'status': 'success'}

@app.route('/template')
def template_response():
    """View rendering template"""
    return render_template('index.html', title='Home')

Application Structure

Small Application Structure

For simple applications with few routes:

1
2
3
4
5
6
7
8
9
10
11
my_flask_app/
β”œβ”€β”€ app.py              # Main application file
β”œβ”€β”€ templates/          # HTML templates
β”‚   β”œβ”€β”€ base.html
β”‚   └── index.html
β”œβ”€β”€ static/             # Static files
β”‚   β”œβ”€β”€ css/
β”‚   β”œβ”€β”€ js/
β”‚   └── images/
β”œβ”€β”€ requirements.txt    # Dependencies
└── .env               # Environment variables

Large Application Structure

For production applications with multiple components:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
my_flask_app/
β”œβ”€β”€ app/
β”‚   β”œβ”€β”€ __init__.py           # Application factory
β”‚   β”œβ”€β”€ models.py             # Database models
β”‚   β”œβ”€β”€ auth/                 # Authentication blueprint
β”‚   β”‚   β”œβ”€β”€ __init__.py
β”‚   β”‚   β”œβ”€β”€ routes.py
β”‚   β”‚   └── forms.py
β”‚   β”œβ”€β”€ blog/                 # Blog blueprint
β”‚   β”‚   β”œβ”€β”€ __init__.py
β”‚   β”‚   β”œβ”€β”€ routes.py
β”‚   β”‚   └── forms.py
β”‚   β”œβ”€β”€ api/                  # API blueprint
β”‚   β”‚   β”œβ”€β”€ __init__.py
β”‚   β”‚   └── routes.py
β”‚   β”œβ”€β”€ templates/            # Templates organized by blueprint
β”‚   β”‚   β”œβ”€β”€ base.html
β”‚   β”‚   β”œβ”€β”€ auth/
β”‚   β”‚   └── blog/
β”‚   β”œβ”€β”€ static/               # Static files
β”‚   β”‚   β”œβ”€β”€ css/
β”‚   β”‚   β”œβ”€β”€ js/
β”‚   β”‚   └── images/
β”‚   └── utils/                # Utility functions
β”‚       β”œβ”€β”€ __init__.py
β”‚       └── helpers.py
β”œβ”€β”€ migrations/               # Database migrations
β”œβ”€β”€ tests/                    # Test files
β”‚   β”œβ”€β”€ __init__.py
β”‚   β”œβ”€β”€ test_auth.py
β”‚   └── test_blog.py
β”œβ”€β”€ instance/                 # Instance-specific files (not in version control)
β”‚   └── config.py
β”œβ”€β”€ config.py                 # Configuration classes
β”œβ”€β”€ requirements.txt          # Production dependencies
β”œβ”€β”€ requirements-dev.txt      # Development dependencies
β”œβ”€β”€ .env                      # Environment variables (not in version control)
β”œβ”€β”€ .gitignore
└── run.py                    # Application entry point

Application Factory Pattern

The application factory pattern creates the Flask application inside a function, enabling multiple app instances with different configurations. This is crucial for testing, deployment flexibility, and avoiding circular imports.

Why Use Application Factory?

Benefits

  1. Multiple Instances: Create apps with different configurations
  2. Testing: Easily create test instances with test configurations
  3. Avoid Circular Imports: Extensions initialized before routes
  4. Configuration Flexibility: Dynamic configuration based on environment
  5. Extension Management: Clean initialization of Flask extensions

Basic Application Factory

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from config import config

# Initialize extensions
db = SQLAlchemy()
migrate = Migrate()

def create_app(config_name='development'):
    """
    Application factory function.
    
    Args:
        config_name: Configuration to use (development, testing, production)
        
    Returns:
        Configured Flask application instance
    """
    app = Flask(__name__)
    
    # Load configuration
    app.config.from_object(config[config_name])
    
    # Initialize extensions with app
    db.init_app(app)
    migrate.init_app(app, db)
    
    # Register blueprints
    from app.auth import auth_bp
    from app.blog import blog_bp
    from app.api import api_bp
    
    app.register_blueprint(auth_bp, url_prefix='/auth')
    app.register_blueprint(blog_bp, url_prefix='/blog')
    app.register_blueprint(api_bp, url_prefix='/api')
    
    # Register error handlers
    register_error_handlers(app)
    
    # Shell context for flask cli
    @app.shell_context_processor
    def make_shell_context():
        return {'db': db, 'User': User, 'Post': Post}
    
    return app

def register_error_handlers(app):
    """Register custom error handlers"""
    
    @app.errorhandler(404)
    def not_found(error):
        return {'error': 'Not found'}, 404
    
    @app.errorhandler(500)
    def internal_error(error):
        db.session.rollback()
        return {'error': 'Internal server error'}, 500

Configuration File

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
# config.py
import os
from datetime import timedelta

basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    """Base configuration"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_RECORD_QUERIES = True
    
    # Session configuration
    PERMANENT_SESSION_LIFETIME = timedelta(days=7)
    SESSION_COOKIE_SECURE = True  # HTTPS only
    SESSION_COOKIE_HTTPONLY = True  # Prevent JavaScript access
    SESSION_COOKIE_SAMESITE = 'Lax'  # CSRF protection
    
    # Upload configuration
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16MB max file size
    UPLOAD_FOLDER = os.path.join(basedir, 'uploads')
    
    @staticmethod
    def init_app(app):
        """Initialize application configuration"""
        pass

class DevelopmentConfig(Config):
    """Development configuration"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'dev-db.sqlite')
    SQLALCHEMY_ECHO = True  # Log SQL queries

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') or \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')
    
    @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
}

Application Entry Point

1
2
3
4
5
6
7
8
9
10
11
# run.py
import os
from app import create_app, db
from app.models import User, Post

# Determine configuration from environment
config_name = os.getenv('FLASK_CONFIG', 'development')
app = create_app(config_name)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Blueprints

Blueprints organize Flask applications into modular components. They group related views, templates, and static files into reusable packages.

Creating a Blueprint

1
2
3
4
5
6
7
8
9
10
11
12
# app/auth/__init__.py
from flask import Blueprint

auth_bp = Blueprint(
    'auth',                          # Blueprint name
    __name__,                        # Import name
    template_folder='templates',     # Blueprint-specific templates
    static_folder='static',          # Blueprint-specific static files
    url_prefix='/auth'               # URL prefix for all routes
)

from app.auth import routes

Blueprint Routes

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# app/auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required
from app.auth import auth_bp
from app.models import User
from app import db

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    """User registration"""
    if request.method == 'POST':
        username = request.form.get('username')
        email = request.form.get('email')
        password = request.form.get('password')
        
        # Check if user exists
        if User.query.filter_by(email=email).first():
            flash('Email already registered', 'error')
            return redirect(url_for('auth.register'))
        
        # Create new user
        user = User(username=username, email=email)
        user.set_password(password)
        db.session.add(user)
        db.session.commit()
        
        flash('Registration successful! Please log in.', 'success')
        return redirect(url_for('auth.login'))
    
    return render_template('auth/register.html')

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    """User login"""
    if request.method == 'POST':
        email = request.form.get('email')
        password = request.form.get('password')
        remember = request.form.get('remember', False)
        
        user = User.query.filter_by(email=email).first()
        
        if user and user.check_password(password):
            login_user(user, remember=remember)
            next_page = request.args.get('next')
            return redirect(next_page or url_for('blog.index'))
        
        flash('Invalid email or password', 'error')
    
    return render_template('auth/login.html')

@auth_bp.route('/logout')
@login_required
def logout():
    """User logout"""
    logout_user()
    flash('You have been logged out', 'success')
    return redirect(url_for('auth.login'))

Registering Blueprints

1
2
3
4
5
6
7
8
9
10
11
12
# In app/__init__.py create_app() function
from app.auth import auth_bp
from app.blog import blog_bp

# Register without prefix
app.register_blueprint(auth_bp)

# Register with prefix
app.register_blueprint(blog_bp, url_prefix='/blog')

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

Blueprint URL Building

1
2
3
4
5
6
7
8
9
# Reference routes from same blueprint
url_for('.index')              # Relative to current blueprint
url_for('auth.login')          # Absolute reference

# Reference routes from different blueprint
url_for('blog.post', post_id=1)

# Reference static files
url_for('auth.static', filename='style.css')

Nested Blueprints

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# app/api/__init__.py
from flask import Blueprint

api_bp = Blueprint('api', __name__, url_prefix='/api')

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

v1_bp = Blueprint('v1', __name__)

# app/api/v1/routes.py
from app.api.v1 import v1_bp

@v1_bp.route('/users')
def get_users():
    return {'users': []}

# Register nested blueprint
# In app/api/__init__.py
from app.api.v1 import v1_bp
api_bp.register_blueprint(v1_bp, url_prefix='/v1')

# In app/__init__.py
app.register_blueprint(api_bp)
# Routes accessible at /api/v1/users

Configuration Management

Environment Variables

Use environment variables for sensitive configuration.

1
2
3
4
5
6
# .env file (never commit to version control)
SECRET_KEY=your-secret-key-here
DATABASE_URL=postgresql://user:pass@localhost/dbname
MAIL_USERNAME=your-email@example.com
MAIL_PASSWORD=your-email-password
FLASK_CONFIG=production

Loading Environment Variables

1
2
3
4
5
6
7
8
9
10
11
# config.py or app/__init__.py
import os
from dotenv import load_dotenv

# Load .env file
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))

# Access environment variables
SECRET_KEY = os.environ.get('SECRET_KEY')
DATABASE_URL = os.environ.get('DATABASE_URL')

Instance Folder

Store instance-specific configuration in the instance/ folder (excluded from version control).

1
2
3
# Instance-relative configuration
app = Flask(__name__, instance_relative_config=True)
app.config.from_pyfile('config.py', silent=True)
1
2
3
# instance/config.py
SECRET_KEY = 'instance-specific-secret'
SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/production_db'

Database Integration

Flask-SQLAlchemy Setup

Flask-SQLAlchemy is the recommended ORM for Flask applications.

1
2
3
4
5
6
7
8
9
10
11
12
# Install
pip install flask-sqlalchemy

# Initialize
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    db.init_app(app)
    return app

Model Definition

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
# app/models.py
from app import db
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    """User model"""
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False, index=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    is_active = db.Column(db.Boolean, default=True)
    
    # Relationships
    posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
    
    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,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat()
        }

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

class Post(db.Model):
    __tablename__ = 'posts'
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    body = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    
    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'body': self.body,
            'created_at': self.created_at.isoformat(),
            'author': self.author.username
        }

# app/auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required
from app.auth import auth_bp
from app.models import User
from app import db

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form.get('username')
        email = request.form.get('email')
        password = request.form.get('password')
        
        if User.query.filter_by(email=email).first():
            flash('Email already registered', 'error')
            return redirect(url_for('auth.register'))
        
        user = User(username=username, email=email)
        user.set_password(password)
        db.session.add(user)
        db.session.commit()
        
        flash('Registration successful!', 'success')
        return redirect(url_for('auth.login'))
    
    return render_template('auth/register.html')

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        email = request.form.get('email')
        password = request.form.get('password')
        
        user = User.query.filter_by(email=email).first()
        
        if user and user.check_password(password):
            login_user(user, remember=True)
            next_page = request.args.get('next')
            return redirect(next_page or url_for('main.index'))
        
        flash('Invalid email or password', 'error')
    
    return render_template('auth/login.html')

@auth_bp.route('/logout')
@login_required
def logout():
    logout_user()
    flash('Logged out successfully', 'success')
    return redirect(url_for('main.index'))

# app/main/routes.py
from flask import render_template, request
from flask_login import login_required, current_user
from app.main import main_bp
from app.models import Post

@main_bp.route('/')
def index():
    page = request.args.get('page', 1, type=int)
    posts = Post.query.order_by(Post.created_at.desc()).paginate(
        page=page, per_page=10, error_out=False
    )
    return render_template('index.html', posts=posts)

@main_bp.route('/dashboard')
@login_required
def dashboard():
    return render_template('dashboard.html', user=current_user)

# app/api/routes.py
from flask import jsonify, request
from flask_login import login_required, current_user
from app.api import api_bp
from app.models import User, Post
from app 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('/posts', methods=['GET'])
def get_posts():
    posts = Post.query.order_by(Post.created_at.desc()).all()
    return jsonify([post.to_dict() for post in posts])

@api_bp.route('/posts', methods=['POST'])
@login_required
def create_post():
    data = request.get_json()
    
    if not data or not data.get('title') or not data.get('body'):
        return jsonify({'error': 'Missing required fields'}), 400
    
    post = Post(
        title=data['title'],
        body=data['body'],
        user_id=current_user.id
    )
    
    db.session.add(post)
    db.session.commit()
    
    return jsonify(post.to_dict()), 201

# run.py
import os
from app import create_app, db
from app.models import User, Post

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

@app.shell_context_processor
def make_shell_context():
    return {'db': db, 'User': User, 'Post': Post}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

References


Conclusion

Flask is a powerful and flexible web framework that provides the foundation for building everything from simple websites to complex web applications and RESTful APIs. Its minimalist design philosophy gives developers the freedom to choose their tools while providing essential functionality out of the box.

Key Takeaways

  1. Start Simple: Flask’s micro-framework nature makes it perfect for beginners and prototyping
  2. Scale Thoughtfully: Use application factory and blueprints for larger applications
  3. Security First: Always implement CSRF protection, secure passwords, and validate input
  4. Test Everything: Write comprehensive tests for models, routes, and APIs
  5. Monitor Performance: Use caching, optimize queries, and implement proper logging
  6. Deploy Carefully: Use production-grade WSGI servers and proper configuration

Next Steps

  • Practice: Build small projects to understand core concepts
  • Read Documentation: Flask’s documentation is excellent and comprehensive
  • Study Extensions: Learn popular Flask extensions for common tasks
  • Join Community: Engage with Flask community on GitHub, Stack Overflow, and Discord
  • Build Real Projects: Apply knowledge to real-world applications

Flask empowers developers to build web applications their way, making it an invaluable tool in any Python developer’s toolkit.

</div>self, password): β€œ"”Hash and set password””” self.password_hash = generate_password_hash(password)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def check_password(self, password):
    """Verify password"""
    return check_password_hash(self.password_hash, password)

def to_dict(self):
    """Serialize to dictionary"""
    return {
        'id': self.id,
        'username': self.username,
        'email': self.email,
        'created_at': self.created_at.isoformat(),
        'is_active': self.is_active
    }

def __repr__(self):
    return f'<User {self.username}>'

class Post(db.Model): β€œ"”Blog post model””” tablename = β€˜posts’

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
slug = db.Column(db.String(200), unique=True, nullable=False, index=True)
body = db.Column(db.Text, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
published = db.Column(db.Boolean, default=False)

# Foreign key
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)

# Many-to-many with tags
tags = db.relationship('Tag', secondary='post_tags', backref=db.backref('posts', lazy='dynamic'))

def __repr__(self):
    return f'<Post {self.title}>'

Association table for many-to-many

post_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 Tag(db.Model): β€œ"”Tag model””” tablename = β€˜tags’

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

Database Migrations with Flask-Migrate

Flask-Migrate uses Alembic to handle database schema changes.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# Install
pip install flask-migrate

# 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
1
2
3
4
5
6
7
8
9
10
11
# Setup in app/__init__.py
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

def create_app():
    app = Flask(__name__)
    db.init_app(app)
    migrate.init_app(app, db)
    return app

Common Database Operations

Create

1
2
3
4
5
6
7
8
9
10
11
12
13
# Create single record
user = User(username='john', email='john@example.com')
user.set_password('password123')
db.session.add(user)
db.session.commit()

# Bulk insert
users = [
    User(username='alice', email='alice@example.com'),
    User(username='bob', email='bob@example.com')
]
db.session.bulk_save_objects(users)
db.session.commit()

Read

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# Get by primary key
user = User.query.get(1)
user = db.session.get(User, 1)  # SQLAlchemy 2.0 style

# Get first or 404
user = User.query.get_or_404(1)

# Filter
users = User.query.filter_by(is_active=True).all()
users = User.query.filter(User.email.like('%@example.com')).all()

# Get one
user = User.query.filter_by(username='john').first()
user = User.query.filter_by(username='john').one_or_none()

# Count
count = User.query.count()
count = User.query.filter_by(is_active=True).count()

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

Update

1
2
3
4
5
6
7
8
# Update single record
user = User.query.get(1)
user.email = 'newemail@example.com'
db.session.commit()

# Update multiple records
User.query.filter_by(is_active=False).update({'is_active': True})
db.session.commit()

Delete

1
2
3
4
5
6
7
8
# Delete single record
user = User.query.get(1)
db.session.delete(user)
db.session.commit()

# Delete multiple records
User.query.filter(User.created_at < datetime(2020, 1, 1)).delete()
db.session.commit()

Advanced Queries

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# Joins
posts = db.session.query(Post).join(User).filter(User.username == 'john').all()

# Eager loading (solve N+1 problem)
from sqlalchemy.orm import joinedload
users = User.query.options(joinedload(User.posts)).all()

# Subqueries
from sqlalchemy import func
subq = db.session.query(
    Post.user_id,
    func.count(Post.id).label('post_count')
).group_by(Post.user_id).subquery()

users_with_counts = db.session.query(User, subq.c.post_count).join(
    subq, User.id == subq.c.user_id
).all()

# Raw SQL
results = db.session.execute(db.text('SELECT * FROM users WHERE id = :id'), {'id': 1})

Request Handling

URL Routing

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# Basic route
@app.route('/')
def index():
    return 'Hello, World!'

# Multiple HTTP methods
@app.route('/submit', methods=['GET', 'POST'])
def submit():
    if request.method == 'POST':
        # Process form
        return redirect(url_for('success'))
    return render_template('form.html')

# Variable rules
@app.route('/user/<username>')
def show_user(username):
    return f'User: {username}'

@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'Post: {post_id}'

# Converters: string (default), int, float, path, uuid
@app.route('/files/<path:filepath>')
def show_file(filepath):
    return f'File: {filepath}'

# Optional segments
@app.route('/posts/')
@app.route('/posts/<int:page>')
def posts(page=1):
    return f'Page: {page}'

Request Object

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from flask import request

@app.route('/data', methods=['POST'])
def handle_data():
    # Form data
    username = request.form.get('username')
    password = request.form['password']  # Raises KeyError if missing
    
    # Query parameters
    page = request.args.get('page', 1, type=int)
    search = request.args.get('q', '')
    
    # JSON data
    data = request.get_json()
    name = data.get('name')
    
    # Files
    file = request.files.get('file')
    if file:
        filename = secure_filename(file.filename)
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    
    # Headers
    user_agent = request.headers.get('User-Agent')
    auth_token = request.headers.get('Authorization')
    
    # Cookies
    session_id = request.cookies.get('session_id')
    
    # Request metadata
    method = request.method
    url = request.url
    base_url = request.base_url
    remote_addr = request.remote_addr
    
    return {'status': 'success'}

Response Handling

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from flask import make_response, jsonify, redirect, url_for, abort

# String response (default 200 status)
@app.route('/')
def index():
    return 'Hello, World!'

# JSON response
@app.route('/api/data')
def api_data():
    return jsonify({'key': 'value', 'numbers': [1, 2, 3]})
    # Or return dict directly (Flask 1.1+)
    return {'key': 'value', 'numbers': [1, 2, 3]}

# Custom status code
@app.route('/created', methods=['POST'])
def create():
    # Method 1: Tuple
    return {'id': 123}, 201
    
    # Method 2: Explicit
    response = jsonify({'id': 123})
    response.status_code = 201
    return response

# Custom headers
@app.route('/custom')
def custom():
    return 'Content', 200, {'X-Custom-Header': 'Value'}

# Response object
@app.route('/cookie')
def set_cookie():
    response = make_response('Cookie set')
    response.set_cookie('user_id', '123', max_age=3600)
    return response

# Redirect
@app.route('/old-page')
def old_page():
    return redirect(url_for('new_page'))

# Abort with error
@app.route('/admin')
def admin():
    if not current_user.is_admin:
        abort(403)  # Forbidden
    return 'Admin panel'

Error Handling

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# Error handlers
@app.errorhandler(404)
def not_found(error):
    return render_template('404.html'), 404

@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()
    return render_template('500.html'), 500

# Custom exception
class ValidationError(Exception):
    """Custom validation exception"""
    pass

@app.errorhandler(ValidationError)
def handle_validation_error(error):
    return {'error': str(error)}, 400

# Using abort
from flask import abort

@app.route('/user/<int:user_id>')
def get_user(user_id):
    user = User.query.get(user_id)
    if not user:
        abort(404, description="User not found")
    return user.to_dict()

Before/After Request Hooks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@app.before_request
def before_request():
    """Runs before each request"""
    g.start_time = time.time()
    
    # Authentication check
    if request.endpoint and 'admin' in request.endpoint:
        if not current_user.is_admin:
            abort(403)

@app.after_request
def after_request(response):
    """Runs after each request"""
    # Add security headers
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    
    # Log request time
    if hasattr(g, 'start_time'):
        elapsed = time.time() - g.start_time
        app.logger.info(f'Request completed in {elapsed:.3f}s')
    
    return response

@app.teardown_request
def teardown_request(exception=None):
    """Runs after request, even if exception occurred"""
    if exception:
        db.session.rollback()

@app.before_first_request
def before_first_request():
    """Runs only before the first request"""
    # Initialize resources
    pass

Templates and Static Files

Jinja2 Templates

Flask uses Jinja2 as its template engine.

Template Inheritance

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
<!-- templates/base.html -->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{% block title %}{% endblock %} - My App</title>
    <link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
    {% block extra_css %}{% endblock %}
</head>
<body>
    <nav>
        <ul>
            <li><a href="{{ url_for('index') }}">Home</a></li>
            {% if current_user.is_authenticated %}
                <li><a href="{{ url_for('auth.logout') }}">Logout</a></li>
            {% else %}
                <li><a href="{{ url_for('auth.login') }}">Login</a></li>
            {% endif %}
        </ul>
    </nav>
    
    <main>
        {% with messages = get_flashed_messages(with_categories=true) %}
            {% if messages %}
                {% for category, message in messages %}
                    <div class="alert alert-{{ category }}">{{ message }}</div>
                {% endfor %}
            {% endif %}
        {% endwith %}
        
        {% block content %}{% endblock %}
    </main>
    
    <footer>
        <p>&copy; 2025 My App</p>
    </footer>
    
    <script src="{{ url_for('static', filename='js/main.js') }}"></script>
    {% block extra_js %}{% endblock %}
</body>
</html>

<!-- templates/index.html -->
{% extends "base.html" %}

{% block title %}Home{% endblock %}

{% block content %}
<h1>Welcome {{ user.username }}!</h1>

{% if posts %}
    {% for post in posts %}
    <article>
        <h2><a href="{{ url_for('blog.post', post_id=post.id) }}">{{ post.title }}</a></h2>
        <p class="meta">By {{ post.author.username }} on {{ post.created_at.strftime('%B %d, %Y') }}</p>
        <p>{{ post.body|truncate(200) }}</p>
    </article>
    {% endfor %}
{% else %}
    <p>No posts yet.</p>
{% endif %}
{% endblock %}

Template Filters

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# Built-in filters
{{ name|upper }}                          # UPPERCASE
{{ name|lower }}                          # lowercase
{{ name|title }}                          # Title Case
{{ text|truncate(100) }}                  # Truncate to 100 chars
{{ text|safe }}                           # Mark as safe HTML
{{ text|escape }}                         # Escape HTML
{{ number|round(2) }}                     # Round to 2 decimals
{{ date|default('N/A') }}                 # Default if undefined
{{ items|length }}                        # Length of list
{{ items|first }}                         # First item
{{ items|last }}                          # Last item
{{ items|join(', ') }}                    # Join with separator

# Custom filters
@app.template_filter('datetime')
def format_datetime(value, format='%Y-%m-%d %H:%M'):
    """Format datetime object"""
    if value is None:
        return ''
    return value.strftime(format)

# Use in template
{{ post.created_at|datetime('%B %d, %Y') }}

Template Context

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# Inject variables into all templates
@app.context_processor
def inject_now():
    return {'now': datetime.utcnow()}

# Use in template
<p>Current time: {{ now.strftime('%H:%M') }}</p>

# Template globals
@app.template_global()
def current_year():
    return datetime.utcnow().year

# Use in template
<footer>&copy; {{ current_year() }} My Company</footer>

Template Macros

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
<!-- templates/macros.html -->
{% macro render_field(field) %}
<div class="form-group">
    {{ field.label }}
    {{ field(class="form-control", **kwargs) }}
    {% if field.errors %}
        {% for error in field.errors %}
            <span class="error">{{ error }}</span>
        {% endfor %}
    {% endif %}
</div>
{% endmacro %}

<!-- templates/form.html -->
{% from "macros.html" import render_field %}

<form method="POST">
    {{ form.hidden_tag() }}
    {{ render_field(form.username) }}
    {{ render_field(form.email) }}
    {{ render_field(form.password) }}
    <button type="submit">Submit</button>
</form>

Static Files

1
2
3
4
5
6
7
8
9
10
11
12
13
# Serve static files
# Files in static/ directory are automatically served

# Reference in templates
<link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
<script src="{{ url_for('static', filename='js/app.js') }}"></script>
<img src="{{ url_for('static', filename='images/logo.png') }}" alt="Logo">

# Cache busting
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0  # Development only

# Or use query string
<link rel="stylesheet" href="{{ url_for('static', filename='css/style.css', v='1.0') }}">

REST API Development

Basic REST API

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
from flask import jsonify, request, abort

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

@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    """Get single user"""
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

@app.route('/api/users', methods=['POST'])
def create_user():
    """Create new user"""
    data = request.get_json()
    
    # Validation
    if not data or not data.get('username') or not data.get('email'):
        abort(400, description="Missing required fields")
    
    # Check if user exists
    if User.query.filter_by(email=data['email']).first():
        abort(409, description="Email already exists")
    
    user = User(
        username=data['username'],
        email=data['email']
    )
    user.set_password(data['password'])
    
    db.session.add(user)
    db.session.commit()
    
    return jsonify(user.to_dict()), 201

@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    """Update user"""
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    
    if 'username' in data:
        user.username = data['username']
    if 'email' in data:
        user.email = data['email']
    
    db.session.commit()
    return jsonify(user.to_dict())

@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    """Delete user"""
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return '', 204

RESTful API Structure

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# app/api/resources.py
from flask import request, jsonify
from flask.views import MethodView
from app.models import User
from app import db

class UserAPI(MethodView):
    """RESTful API for User resource"""
    
    def get(self, user_id):
        """GET /api/users/<id>"""
        if user_id is None:
            # List all users
            page = request.args.get('page', 1, type=int)
            per_page = request.args.get('per_page', 20, type=int)
            
            pagination = User.query.paginate(
                page=page, 
                per_page=per_page, 
                error_out=False
            )
            
            return jsonify({
                'users': [user.to_dict() for user in pagination.items],
                'total': pagination.total,
                'page': page,
                'per_page': per_page,
                'pages': pagination.pages
            })
        else:
            # Get single user
            user = User.query.get_or_404(user_id)
            return jsonify(user.to_dict())
    
    def post(self):
        """POST /api/users"""
        data = request.get_json()
        
        # Validation
        required_fields = ['username', 'email', 'password']
        if not all(field in data for field in required_fields):
            return jsonify({'error': 'Missing required fields'}), 400
        
        user = User(username=data['username'], email=data['email'])
        user.set_password(data['password'])
        
        db.session.add(user)
        db.session.commit()
        
        return jsonify(user.to_dict()), 201
    
    def put(self, user_id):
        """PUT /api/users/<id>"""
        user = User.query.get_or_404(user_id)
        data = request.get_json()
        
        user.username = data.get('username', user.username)
        user.email = data.get('email', user.email)
        
        db.session.commit()
        return jsonify(user.to_dict())
    
    def delete(self, user_id):
        """DELETE /api/users/<id>"""
        user = User.query.get_or_404(user_id)
        db.session.delete(user)
        db.session.commit()
        return '', 204

# Register view
user_view = UserAPI.as_view('user_api')
app.add_url_rule('/api/users', defaults={'user_id': None},
                 view_func=user_view, methods=['GET'])
app.add_url_rule('/api/users', view_func=user_view, methods=['POST'])
app.add_url_rule('/api/users/<int:user_id>', view_func=user_view,
                 methods=['GET', 'PUT', 'DELETE'])

Flask-RESTful Extension

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# Install
pip install flask-restful

# Setup
from flask_restful import Resource, Api, reqparse, fields, marshal_with

api = Api(app)

# Define output fields
user_fields = {
    'id': fields.Integer,
    'username': fields.String,
    'email': fields.String,
    'created_at': fields.DateTime(dt_format='iso8601')
}

class UserResource(Resource):
    @marshal_with(user_fields)
    def get(self, user_id):
        """Get user"""
        user = User.query.get_or_404(user_id)
        return user
    
    def post(self):
        """Create user"""
        parser = reqparse.RequestParser()
        parser.add_argument('username', required=True, help='Username required')
        parser.add_argument('email', required=True, help='Email required')
        parser.add_argument('password', required=True, help='Password required')
        args = parser.parse_args()
        
        user = User(username=args['username'], email=args['email'])
        user.set_password(args['password'])
        
        db.session.add(user)
        db.session.commit()
        
        return user.to_dict(), 201

class UserListResource(Resource):
    @marshal_with(user_fields)
    def get(self):
        """List users"""
        return User.query.all()

# Register resources
api.add_resource(UserListResource, '/api/users')
api.add_resource(UserResource, '/api/users/<int:user_id>')

API Versioning

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# URL versioning
@app.route('/api/v1/users')
def api_v1_users():
    return jsonify({'version': 'v1', 'users': []})

@app.route('/api/v2/users')
def api_v2_users():
    return jsonify({'version': 'v2', 'users': []})

# Header versioning
@app.route('/api/users')
def api_users():
    version = request.headers.get('API-Version', 'v1')
    if version == 'v2':
        return jsonify({'version': 'v2', 'users': []})
    return jsonify({'version': 'v1', 'users': []})

CORS (Cross-Origin Resource Sharing)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# Install
pip install flask-cors

# Setup
from flask_cors import CORS

# Enable for all routes
CORS(app)

# Enable for specific routes
CORS(app, resources={r"/api/*": {"origins": "http://localhost:3000"}})

# Configure per route
@app.route('/api/data')
@cross_origin()
def api_data():
    return jsonify({'data': 'value'})

# Advanced configuration
CORS(app, resources={
    r"/api/*": {
        "origins": ["http://localhost:3000", "https://example.com"],
        "methods": ["GET", "POST", "PUT", "DELETE"],
        "allow_headers": ["Content-Type", "Authorization"],
        "expose_headers": ["X-Total-Count"],
        "max_age": 3600
    }
})

Authentication and Authorization

Flask-Login

Flask-Login manages user sessions.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# Install
pip install flask-login

# Setup
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'  # Redirect unauthorized users
login_manager.login_message = 'Please log in to access this page.'

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

# Update User model
class User(UserMixin, db.Model):
    # ... existing fields ...
    
    def get_id(self):
        """Return user ID as string"""
        return str(self.id)
    
    @property
    def is_active(self):
        """Check if user is active"""
        return self.active
    
    @property
    def is_authenticated(self):
        """Check if user is authenticated"""
        return True
    
    @property
    def is_anonymous(self):
        """Check if user is anonymous"""
        return False

Login/Logout

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from flask import flash, redirect, url_for
from flask_login import login_user, logout_user, login_required

@app.route('/login', methods=['POST'])
def login():
    email = request.form.get('email')
    password = request.form.get('password')
    remember = request.form.get('remember', False)
    
    user = User.query.filter_by(email=email).first()
    
    if user and user.check_password(password):
        login_user(user, remember=remember)
        
        # Redirect to next page or home
        next_page = request.args.get('next')
        if not next_page or not next_page.startswith('/'):
            next_page = url_for('index')
        
        return redirect(next_page)
    
    flash('Invalid email or password', 'error')
    return redirect(url_for('auth.login'))

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('index'))

# Protected route
@app.route('/dashboard')
@login_required
def dashboard():
    return f'Welcome, {current_user.username}!'

Role-Based Access Control

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from functools import wraps
from flask import abort
from flask_login import current_user

# Role model
class Role(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True)
    
    users = db.relationship('User', backref='role', lazy='dynamic')

class User(UserMixin, db.Model):
    # ... existing fields ...
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    
    def has_role(self, role_name):
        """Check if user has specific role"""
        return self.role and self.role.name == role_name

# Role decorator
def role_required(role_name):
    """Decorator to require specific role"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.is_authenticated:
                abort(401)
            if not current_user.has_role(role_name):
                abort(403)
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# Usage
@app.route('/admin')
@login_required
@role_required('admin')
def admin_panel():
    return 'Admin Panel'

JWT Authentication (API)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
# Install
pip install pyjwt

# Token generation
import jwt
from datetime import datetime, timedelta
from flask import current_app

def generate_token(user_id):
    """Generate JWT token"""
    payload = {
        'user_id': user_id,
        'exp': datetime.utcnow() + timedelta(days=1),
        'iat': datetime.utcnow()
    }
    token = jwt.encode(payload, current_app.config['SECRET_KEY'], algorithm='HS256')
    return token

def verify_token(token):
    """Verify JWT token"""
    try:
        payload = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
        return payload['user_id']
    except jwt.ExpiredSignatureError:
        return None  # Token expired
    except jwt.InvalidTokenError:
        return None  # Invalid token

# Token authentication decorator
from functools import wraps

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization')
        
        if not token:
            return jsonify({'error': 'Token missing'}), 401
        
        if token.startswith('Bearer '):
            token = token[7:]
        
        user_id = verify_token(token)
        if not user_id:
            return jsonify({'error': 'Invalid or expired token'}), 401
        
        # Get user
        user = User.query.get(user_id)
        if not user:
            return jsonify({'error': 'User not found'}), 401
        
        return f(user, *args, **kwargs)
    return decorated

# Protected API endpoint
@app.route('/api/protected')
@token_required
def protected(user):
    return jsonify({'message': f'Hello, {user.username}!'})

# Login endpoint
@app.route('/api/login', methods=['POST'])
def api_login():
    data = request.get_json()
    user = User.query.filter_by(email=data['email']).first()
    
    if user and user.check_password(data['password']):
        token = generate_token(user.id)
        return jsonify({'token': token, 'user': user.to_dict()})
    
    return jsonify({'error': 'Invalid credentials'}), 401

Security Best Practices

Secret Key Management

1
2
3
4
5
6
7
8
9
10
# Generate strong secret key
import secrets
secret_key = secrets.token_hex(32)

# Store in environment variable
# .env
SECRET_KEY=your-generated-secret-key

# Never hardcode in source code
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')

Password Hashing

1
2
3
4
5
6
7
8
9
10
11
12
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    password_hash = db.Column(db.String(255))
    
    def set_password(self, password):
        """Hash password before storing"""
        self.password_hash = generate_password_hash(password, method='pbkdf2:sha256')
    
    def check_password(self, password):
        """Verify password"""
        return check_password_hash(self.password_hash, password)

CSRF Protection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# Install
pip install flask-wtf

# Setup
from flask_wtf.csrf import CSRFProtect

csrf = CSRFProtect()
csrf.init_app(app)

# In forms
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField

class LoginForm(FlaskForm):
    username = StringField('Username')
    password = PasswordField('Password')

# In template
<form method="POST">
    {{ form.hidden_tag() }}  {# Includes CSRF token #}
    {{ form.username }}
    {{ form.password }}
    <button type="submit">Login</button>
</form>

# Exempt API routes
@csrf.exempt
@app.route('/api/webhook', methods=['POST'])
def webhook():
    return jsonify({'status': 'received'})

Security Headers

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@app.after_request
def set_security_headers(response):
    """Add security headers to all responses"""
    # Prevent MIME type sniffing
    response.headers['X-Content-Type-Options'] = 'nosniff'
    
    # Prevent clickjacking
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    
    # Enable XSS protection
    response.headers['X-XSS-Protection'] = '1; mode=block'
    
    # Content Security Policy
    response.headers['Content-Security-Policy'] = "default-src 'self'"
    
    # HTTPS only
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    
    return response

SQL Injection Prevention

1
2
3
4
5
6
7
8
9
10
# SAFE: Using SQLAlchemy ORM (parameterized queries)
user = User.query.filter_by(username=username).first()

# SAFE: Using text() with parameters
from sqlalchemy import text
result = db.session.execute(text('SELECT * FROM users WHERE username = :username'), 
                           {'username': username})

# UNSAFE: String concatenation (DO NOT USE)
# query = f"SELECT * FROM users WHERE username = '{username}'"  # VULNERABLE!

File Upload Security

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from werkzeug.utils import secure_filename
import os

ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}

def allowed_file(filename):
    """Check if file extension is allowed"""
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return {'error': 'No file provided'}, 400
    
    file = request.files['file']
    
    if file.filename == '':
        return {'error': 'Empty filename'}, 400
    
    if not allowed_file(file.filename):
        return {'error': 'File type not allowed'}, 400
    
    # Secure filename
    filename = secure_filename(file.filename)
    
    # Save with unique name
    unique_filename = f"{uuid.uuid4()}_{filename}"
    filepath = os.path.join(app.config['UPLOAD_FOLDER'], unique_filename)
    file.save(filepath)
    
    return {'filename': unique_filename}, 201

Rate Limiting

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# Install
pip install flask-limiter

# Setup
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app=app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="redis://localhost:6379"
)

# Apply to routes
@app.route('/api/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    # Login logic
    pass

# Different limits for different routes
@app.route('/api/expensive')
@limiter.limit("10 per hour")
def expensive_operation():
    pass

Testing

Test Structure

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# tests/conftest.py
import pytest
from app import create_app, db
from app.models import User

@pytest.fixture
def app():
    """Create application for testing"""
    app = create_app('testing')
    
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

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

@pytest.fixture
def runner(app):
    """CLI test runner"""
    return app.test_cli_runner()

@pytest.fixture
def user(app):
    """Create test user"""
    user = User(username='testuser', email='test@example.com')
    user.set_password('password123')
    db.session.add(user)
    db.session.commit()
    return user

Unit Tests

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# tests/test_models.py
def test_user_password_hashing(app):
    """Test password hashing"""
    user = User(username='test', email='test@example.com')
    user.set_password('password')
    
    assert user.password_hash is not None
    assert user.password_hash != 'password'
    assert user.check_password('password')
    assert not user.check_password('wrongpassword')

def test_user_to_dict(app, user):
    """Test user serialization"""
    user_dict = user.to_dict()
    
    assert 'id' in user_dict
    assert user_dict['username'] == 'testuser'
    assert user_dict['email'] == 'test@example.com'
    assert 'password_hash' not in user_dict  # Never expose password

Integration Tests

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# tests/test_routes.py
def test_index_page(client):
    """Test index page loads"""
    response = client.get('/')
    assert response.status_code == 200
    assert b'Welcome' in response.data

def test_login_success(client, user):
    """Test successful login"""
    response = client.post('/auth/login', data={
        'email': 'test@example.com',
        'password': 'password123'
    }, follow_redirects=True)
    
    assert response.status_code == 200
    assert b'Welcome' in response.data

def test_login_invalid_password(client, user):
    """Test login with wrong password"""
    response = client.post('/auth/login', data={
        'email': 'test@example.com',
        'password': 'wrongpassword'
    })
    
    assert response.status_code == 200
    assert b'Invalid' in response.data

def test_protected_route_unauthorized(client):
    """Test accessing protected route without login"""
    response = client.get('/dashboard')
    assert response.status_code == 302  # Redirect to login

API Tests

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# tests/test_api.py
def test_get_users(client):
    """Test GET /api/users"""
    response = client.get('/api/users')
    assert response.status_code == 200
    data = response.get_json()
    assert isinstance(data, list)

def test_create_user(client):
    """Test POST /api/users"""
    response = client.post('/api/users', json={
        'username': 'newuser',
        'email': 'new@example.com',
        'password': 'password123'
    })
    
    assert response.status_code == 201
    data = response.get_json()
    assert data['username'] == 'newuser'

def test_get_nonexistent_user(client):
    """Test GET /api/users/999"""
    response = client.get('/api/users/999')
    assert response.status_code == 404

Running Tests

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# Install pytest
pip install pytest pytest-cov

# Run all tests
pytest

# Run with coverage
pytest --cov=app tests/

# Run specific test file
pytest tests/test_models.py

# Run specific test
pytest tests/test_models.py::test_user_password_hashing

# Verbose output
pytest -v

# Stop on first failure
pytest -x

Deployment

Production Configuration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# config.py
class ProductionConfig(Config):
    DEBUG = False
    TESTING = False
    
    # Use environment variables
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
    SECRET_KEY = os.environ.get('SECRET_KEY')
    
    # Security settings
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Lax'
    
    # Logging
    @classmethod
    def init_app(cls, app):
        Config.init_app(app)
        
        # Log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

WSGI Server (Gunicorn)

1
2
3
4
5
6
7
8
# Install
pip install gunicorn

# Run
gunicorn -w 4 -b 0.0.0.0:8000 "app:create_app()"

# With configuration file
gunicorn -c gunicorn_config.py "app:create_app()"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# gunicorn_config.py
workers = 4
worker_class = 'sync'
worker_connections = 1000
timeout = 30
keepalive = 2

bind = '0.0.0.0:8000'

# Logging
accesslog = '/var/log/gunicorn/access.log'
errorlog = '/var/log/gunicorn/error.log'
loglevel = 'info'

# Process naming
proc_name = 'my_flask_app'

# Server mechanics
daemon = False
pidfile = '/var/run/gunicorn.pid'
user = 'www-data'
group = 'www-data'

Docker Deployment

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Create non-root user
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# Expose port
EXPOSE 8000

# Run application
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:create_app()"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLASK_CONFIG=production
      - DATABASE_URL=postgresql://user:pass@db:5432/dbname
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      - db
    volumes:
      - ./uploads:/app/uploads
    restart: unless-stopped
  
  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=dbname
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped
  
  nginx:
    image: nginx:latest
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./static:/usr/share/nginx/html/static
    depends_on:
      - web
    restart: unless-stopped

volumes:
  postgres_data:

Nginx Configuration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# nginx.conf
upstream flask_app {
    server web:8000;
}

server {
    listen 80;
    server_name example.com;
    
    # Redirect to HTTPS
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name example.com;
    
    # SSL configuration
    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;
    
    # Security headers
    add_header Strict-Transport-Security "max-age=31536000" always;
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;
    
    # Static files
    location /static {
        alias /usr/share/nginx/html/static;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }
    
    # Proxy to Flask
    location / {
        proxy_pass http://flask_app;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_redirect off;
        
        # Timeouts
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
    
    # Increase upload size
    client_max_body_size 16M;
}

Environment Variables

1
2
3
4
5
6
7
8
9
# .env.production (never commit)
FLASK_CONFIG=production
SECRET_KEY=your-production-secret-key
DATABASE_URL=postgresql://user:password@localhost/production_db
MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USE_TLS=True
MAIL_USERNAME=your-email@gmail.com
MAIL_PASSWORD=your-app-password

Performance Optimization

Database Query Optimization

N+1 Query Problem

1
2
3
4
5
6
7
8
9
10
11
# INEFFICIENT: N+1 queries
posts = Post.query.all()
for post in posts:
    print(post.author.username)  # Separate query for each author

# EFFICIENT: Use eager loading
from sqlalchemy.orm import joinedload

posts = Post.query.options(joinedload(Post.author)).all()
for post in posts:
    print(post.author.username)  # Author already loaded

Select Only Needed Columns

1
2
3
4
5
# Load only specific columns
users = db.session.query(User.id, User.username).all()

# Or using with_entities
users = User.query.with_entities(User.id, User.username).all()

Pagination

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@app.route('/posts')
def posts():
    page = request.args.get('page', 1, type=int)
    per_page = 20
    
    pagination = Post.query.order_by(Post.created_at.desc()).paginate(
        page=page,
        per_page=per_page,
        error_out=False
    )
    
    return render_template('posts.html',
                         posts=pagination.items,
                         pagination=pagination)

Database Indexing

1
2
3
4
5
6
7
8
9
10
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, index=True)  # Index for lookups
    email = db.Column(db.String(120), unique=True, index=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    
    # Composite index
    __table_args__ = (
        db.Index('idx_username_email', 'username', 'email'),
    )

Caching

Flask-Caching

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# Install
pip install Flask-Caching

# Setup
from flask_caching import Cache

cache = Cache(config={
    'CACHE_TYPE': 'redis',
    'CACHE_REDIS_URL': 'redis://localhost:6379/0',
    'CACHE_DEFAULT_TIMEOUT': 300
})
cache.init_app(app)

# Cache view function
@app.route('/expensive')
@cache.cached(timeout=600)
def expensive_operation():
    # Complex calculation
    result = perform_calculation()
    return jsonify(result)

# Cache with query parameters
@app.route('/posts')
@cache.cached(timeout=300, query_string=True)
def posts():
    page = request.args.get('page', 1, type=int)
    posts = Post.query.paginate(page=page, per_page=20)
    return render_template('posts.html', posts=posts)

# Cache function result
@cache.memoize(timeout=600)
def get_user_posts(user_id):
    return Post.query.filter_by(user_id=user_id).all()

# Manual cache operations
cache.set('key', 'value', timeout=300)
value = cache.get('key')
cache.delete('key')
cache.clear()  # Clear all cache

Asynchronous Tasks

Celery

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# Install
pip install celery redis

# celery_app.py
from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    
    celery.Task = ContextTask
    return celery

# Configuration
app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379/0',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)

celery = make_celery(app)

# Define tasks
@celery.task
def send_email(user_id, subject, body):
    """Send email asynchronously"""
    user = User.query.get(user_id)
    # Send email logic
    return f"Email sent to {user.email}"

@celery.task
def process_data(data_id):
    """Process data in background"""
    # Heavy processing
    return "Processing complete"

# Use in routes
@app.route('/send-email/<int:user_id>')
def trigger_email(user_id):
    send_email.delay(user_id, "Subject", "Body")
    return "Email queued for sending"

# Run Celery worker
# celery -A celery_app.celery worker --loglevel=info

Response Compression

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# Install
pip install flask-compress

# Setup
from flask_compress import Compress

compress = Compress()
compress.init_app(app)

# Configuration
app.config['COMPRESS_MIMETYPES'] = [
    'text/html',
    'text/css',
    'text/javascript',
    'application/json',
    'application/javascript'
]
app.config['COMPRESS_LEVEL'] = 6
app.config['COMPRESS_MIN_SIZE'] = 500

Static File Optimization

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# Use CDN for static files
app.config['CDN_DOMAIN'] = 'cdn.example.com'

@app.context_processor
def override_url_for():
    return dict(url_for=cdn_url_for)

def cdn_url_for(endpoint, **values):
    if endpoint == 'static':
        filename = values.get('filename')
        if filename:
            return f"https://{app.config['CDN_DOMAIN']}/static/{filename}"
    return url_for(endpoint, **values)

# Asset bundling and minification
# Use Flask-Assets
pip install Flask-Assets

from flask_assets import Environment, Bundle

assets = Environment(app)

css = Bundle('css/style.css', 'css/theme.css',
            filters='cssmin', output='gen/packed.css')
assets.register('css_all', css)

js = Bundle('js/app.js', 'js/utils.js',
           filters='jsmin', output='gen/packed.js')
assets.register('js_all', js)

# In template
{% assets "css_all" %}
    <link rel="stylesheet" href="{{ ASSET_URL }}">
{% endassets %}

Best Practices

1. Use Application Factory Pattern

Always use the application factory for scalability and testability.

1
2
3
4
5
6
7
8
9
# Good
def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    # Initialize extensions
    return app

# Bad
app = Flask(__name__)  # Global app instance

2. Organize Code with Blueprints

Modularize large applications using blueprints.

1
2
3
4
5
6
# app/auth/__init__.py
from flask import Blueprint

auth_bp = Blueprint('auth', __name__)

from app.auth import routes

3. Keep Configuration Separate

Never hardcode configuration in source files.

1
2
3
4
5
# Good
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')

# Bad
app.config['SECRET_KEY'] = 'hardcoded-secret'  # NEVER DO THIS

4. Use Environment Variables

Store sensitive data in environment variables.

1
2
3
4
5
from dotenv import load_dotenv
load_dotenv()

DATABASE_URL = os.environ.get('DATABASE_URL')
SECRET_KEY = os.environ.get('SECRET_KEY')

5. Implement Proper Error Handling

Handle errors gracefully with custom error pages.

1
2
3
4
5
6
7
8
9
10
@app.errorhandler(404)
def not_found(error):
    if request.path.startswith('/api/'):
        return jsonify({'error': 'Not found'}), 404
    return render_template('404.html'), 404

@app.errorhandler(Exception)
def handle_exception(error):
    app.logger.error(f'Unhandled exception: {error}')
    return jsonify({'error': 'Internal server error'}), 500

6. Use Database Migrations

Always use migrations for database schema changes.

1
2
3
flask db init
flask db migrate -m "Description of changes"
flask db upgrade

7. Validate Input Data

Always validate and sanitize user input.

1
2
3
4
5
6
7
8
9
10
11
from wtforms import Form, StringField, validators

class RegistrationForm(Form):
    username = StringField('Username', [
        validators.Length(min=4, max=25),
        validators.Regexp('^[a-zA-Z0-9_]+)
    ])
    email = StringField('Email', [
        validators.Email(),
        validators.Length(min=6, max=120)
    ])

8. Use Context Managers for Resources

Properly manage database sessions and file handles.

1
2
3
4
5
6
7
8
9
10
11
12
13
# Good - automatic cleanup
with app.app_context():
    db.session.add(user)
    db.session.commit()

# Better - handle exceptions
try:
    db.session.add(user)
    db.session.commit()
except Exception as e:
    db.session.rollback()
    app.logger.error(f'Error: {e}')
    raise

9. Implement Logging

Use proper logging instead of print statements.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Use in application
@app.route('/process')
def process():
    app.logger.info('Processing started')
    try:
        result = perform_operation()
        app.logger.info('Processing completed')
        return result
    except Exception as e:
        app.logger.error(f'Processing failed: {e}')
        raise

10. Use Dependency Injection

Pass dependencies explicitly for testability.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Good
def create_user(db, username, email):
    user = User(username=username, email=email)
    db.session.add(user)
    db.session.commit()
    return user

# Better for testing
class UserService:
    def __init__(self, db):
        self.db = db
    
    def create_user(self, username, email):
        user = User(username=username, email=email)
        self.db.session.add(user)
        self.db.session.commit()
        return user

11. Implement API Versioning

Version your APIs from the start.

1
2
3
# URL versioning
api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')
api_v2 = Blueprint('api_v2', __name__, url_prefix='/api/v2')

12. Use Pagination for Large Datasets

Never return all records at once.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@app.route('/api/users')
def get_users():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    pagination = User.query.paginate(
        page=page,
        per_page=min(per_page, 100),  # Cap at 100
        error_out=False
    )
    
    return jsonify({
        'users': [u.to_dict() for u in pagination.items],
        'total': pagination.total,
        'pages': pagination.pages,
        'current_page': page
    })

13. Implement Request Timeouts

Set reasonable timeouts for operations.

1
2
3
4
5
6
7
8
from werkzeug.exceptions import RequestTimeout

@app.before_request
def check_timeout():
    request.environ.setdefault('werkzeug.server.shutdown', None)

# Configure gunicorn timeout
# gunicorn --timeout 30 app:app

14. Document Your API

Use tools like Flask-RESTX or Flask-Swagger for API documentation.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# Install
pip install flask-restx

from flask_restx import Api, Resource, fields

api = Api(app, version='1.0', title='My API',
          description='A simple API')

user_model = api.model('User', {
    'id': fields.Integer(readonly=True),
    'username': fields.String(required=True),
    'email': fields.String(required=True)
})

@api.route('/users/<int:id>')
class UserResource(Resource):
    @api.doc('get_user')
    @api.marshal_with(user_model)
    def get(self, id):
        """Get a user by ID"""
        return User.query.get_or_404(id)

15. Monitor Application Performance

Implement application monitoring.

1
2
3
4
5
6
7
# Using Flask-Monitoring-Dashboard
pip install flask-monitoringdashboard

import flask_monitoringdashboard as dashboard
dashboard.bind(app)

# Or use APM tools like New Relic, DataDog

Common Pitfalls

1. Not Using Application Factory

Problem: Hard to test and configure multiple instances.

Solution: Use application factory pattern.

1
2
3
4
5
6
7
8
# Wrong
app = Flask(__name__)

# Right
def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    return app

2. Circular Imports

Problem: Modules importing each other cause import errors.

Solution: Import at function level or restructure code.

1
2
3
4
5
6
7
8
9
10
11
12
13
# Wrong
# app/__init__.py
from app.routes import main_bp  # Circular import

# Right
# app/__init__.py
def create_app():
    app = Flask(__name__)
    
    from app.routes import main_bp  # Import inside function
    app.register_blueprint(main_bp)
    
    return app

3. Not Handling Database Sessions Properly

Problem: Uncommitted sessions or not closing connections.

Solution: Use context managers and explicit commits.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# Wrong
def create_user():
    user = User(username='john')
    db.session.add(user)
    # Forgot commit!

# Right
def create_user():
    user = User(username='john')
    db.session.add(user)
    try:
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        raise

4. Exposing Sensitive Information

Problem: Showing stack traces or internal errors to users.

Solution: Use DEBUG=False in production and custom error pages.

1
2
3
4
5
6
7
8
9
10
11
12
# Production config
class ProductionConfig(Config):
    DEBUG = False
    TESTING = False

# Custom error handler
@app.errorhandler(500)
def internal_error(error):
    # Log the error
    app.logger.error(f'Server Error: {error}')
    # Show generic message
    return render_template('500.html'), 500

5. Not Validating User Input

Problem: Security vulnerabilities and data corruption.

Solution: Always validate and sanitize input.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Wrong
@app.route('/user/<username>')
def get_user(username):
    user = User.query.filter_by(username=username).first()
    # What if username contains SQL injection?

# Right
from wtforms import validators

@app.route('/user/<username>')
def get_user(username):
    # Validate username format
    if not username.isalnum() or len(username) > 50:
        abort(400, 'Invalid username')
    
    user = User.query.filter_by(username=username).first_or_404()
    return user.to_dict()

6. Hardcoding Secrets

Problem: Exposed credentials in version control.

Solution: Use environment variables.

1
2
3
4
5
6
7
# Wrong
app.config['SECRET_KEY'] = 'my-secret-key'
app.config['DATABASE_URL'] = 'postgresql://user:pass@localhost/db'

# Right
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')
app.config['DATABASE_URL'] = os.environ.get('DATABASE_URL')

7. Not Using Blueprints for Large Apps

Problem: Single file becomes unmanageable.

Solution: Organize with blueprints.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# Wrong - everything in one file
@app.route('/auth/login')
@app.route('/auth/register')
@app.route('/blog/posts')
# 100s of routes...

# Right - organized with blueprints
# app/auth/routes.py
@auth_bp.route('/login')
def login():
    pass

# app/blog/routes.py
@blog_bp.route('/posts')
def posts():
    pass

8. Ignoring CSRF Protection

Problem: Vulnerable to cross-site request forgery.

Solution: Enable CSRF protection.

1
2
3
4
5
6
from flask_wtf.csrf import CSRFProtect

csrf = CSRFProtect(app)

# In forms

9. Not Setting Up Proper Logging

Problem: Hard to debug production issues.

Solution: Configure comprehensive logging.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import logging
from logging.handlers import RotatingFileHandler

if not app.debug:
    file_handler = RotatingFileHandler('logs/app.log', 
                                      maxBytes=10240, 
                                      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')

10. Not Using Database Migrations

Problem: Manual schema changes lead to inconsistencies.

Solution: Always use Flask-Migrate.

1
2
3
4
5
6
7
8
# Initialize migrations
flask db init

# Create migration
flask db migrate -m "Add user table"

# Apply migration
flask db upgrade

Jargon Tables

Table 1: Flask Development Lifecycle Terminology

Flask TermAlternative TermsDefinitionContext
Routeendpoint, URL pattern, view mappingURL pattern mapped to a view functionURL routing
View Functionroute handler, controller, endpoint functionFunction that handles HTTP requestsRequest handling
Blueprintmodule, sub-application, componentModular component of Flask applicationApplication structure
Application Factoryapp factory, factory patternFunction that creates Flask application instancesApplication creation
Contextrequest context, application contextEnvironment data available during requestRequest lifecycle
Extensionplugin, add-on, library integrationThird-party library integrated with FlaskFeature enhancement
Templateview, HTML template, Jinja templateHTML file with dynamic content placeholdersFrontend rendering
Static Filesassets, resources, media filesCSS, JavaScript, images, fontsFrontend resources
Migrationschema change, database evolutionDatabase schema version controlDatabase management
WSGIWeb Server Gateway InterfaceStandard interface between web server and appDeployment
Request Objectreq, HTTP requestObject containing request dataRequest handling
Response ObjectHTTP response, replyObject returned to clientResponse handling
Sessionuser session, cookie sessionUser-specific data storageState management
Flash Messagealert, notification, temporary messageOne-time message displayed to userUser feedback
g Objectglobal, request-local storageRequest-local data storageData sharing

Table 2: Hierarchical Application Structure

LevelComponentSub-ComponentPurposeFile/Folder
1Application RootΒ Top-level application/
2Application PackageΒ Main application code/app/
Β Β ModelsDatabase models/app/models.py
Β Β Views/RoutesRequest handlers/app/routes.py
Β Β FormsForm definitions/app/forms.py
Β Β BlueprintsModular components/app/auth/, /app/blog/
2ConfigurationΒ Application settings/config.py
Β Β Development ConfigDev environment settingsClass in config.py
Β Β Production ConfigProd environment settingsClass in config.py
Β Β Testing ConfigTest environment settingsClass in config.py
2TemplatesΒ HTML templates/app/templates/
Β Β Base TemplatesLayout templates/app/templates/base.html
Β Β Blueprint TemplatesComponent-specific templates/app/templates/auth/
2Static FilesΒ Frontend assets/app/static/
Β Β CSSStylesheets/app/static/css/
Β Β JavaScriptClient-side scripts/app/static/js/
Β Β ImagesGraphics and icons/app/static/images/
2MigrationsΒ Database schema versions/migrations/
2TestsΒ Test suite/tests/
Β Β Unit TestsComponent tests/tests/test_models.py
Β Β Integration TestsSystem tests/tests/test_routes.py
2Instance FolderΒ Instance-specific files/instance/

Table 3: Request-Response Cycle Stages

StageTechnical TermDescriptionFlask ComponentsHTTP Layer
1Request InitiationClient sends HTTP requestWSGI server receives requestHTTP Request
2Request ParsingParse HTTP request dataWerkzeug processes requestHeaders, Body, Method
3URL RoutingMatch URL to view functionFlask router, @app.route()URL, Path Parameters
4Before RequestPre-processing hooks@app.before_requestN/A
5Context CreationPush request/app contextsRequest context, App contextN/A
6View ExecutionExecute matched view functionView function, Business logicN/A
7Response CreationGenerate HTTP responsemake_response(), jsonify()Status Code, Headers
8After RequestPost-processing hooks@app.after_requestN/A
9Context TeardownClean up contexts@app.teardown_requestN/A
10Response SendingSend response to clientWSGI server, HTTPHTTP Response

Table 4: Database Operations Terminology

OperationSQL EquivalentSQLAlchemy MethodFlask-SQLAlchemyUse Case
CreateINSERTsession.add()db.session.add()Add new record
ReadSELECTquery.filter()Model.query.filter_by()Retrieve records
UpdateUPDATEModify object + commitModify + db.session.commit()Change existing record
DeleteDELETEsession.delete()db.session.delete()Remove record
QuerySELECTquery()Model.queryBuild query
FilterWHEREfilter(), filter_by().filter_by(field=value)Conditional selection
JoinJOINjoin().join(Model)Combine tables
OrderORDER BYorder_by().order_by(Model.field)Sort results
LimitLIMITlimit().limit(n)Restrict count
CountCOUNT(*)count().count()Count records
PaginationLIMIT + OFFSETpaginate().paginate(page, per_page)Paginated results
CommitCOMMITsession.commit()db.session.commit()Save changes
RollbackROLLBACKsession.rollback()db.session.rollback()Undo changes

Table 5: HTTP Methods and Flask Decorators

HTTP MethodPurposeIdempotentSafeFlask DecoratorTypical Use
GETRetrieve resourceYesYes@app.route('/', methods=['GET'])View data, listings
POSTCreate resourceNoNo@app.route('/', methods=['POST'])Submit forms, create
PUTUpdate/Replace resourceYesNo@app.route('/<id>', methods=['PUT'])Full update
PATCHPartial updateNoNo@app.route('/<id>', methods=['PATCH'])Partial update
DELETEDelete resourceYesNo@app.route('/<id>', methods=['DELETE'])Remove resource
HEADGet headers onlyYesYesAutomatic with GETCheck existence
OPTIONSGet allowed methodsYesYesAutomaticCORS preflight

Table 6: Extension Categories

CategoryExtension NamePurposeInstallationCommon Use
DatabaseFlask-SQLAlchemyORM for databasespip install flask-sqlalchemyDatabase models
Β Flask-MigrateDatabase migrationspip install flask-migrateSchema versioning
AuthenticationFlask-LoginUser session managementpip install flask-loginLogin/logout
Β Flask-JWT-ExtendedJWT authenticationpip install flask-jwt-extendedAPI authentication
FormsFlask-WTFForm handling and CSRFpip install flask-wtfForm validation
APIFlask-RESTfulREST API frameworkpip install flask-restfulRESTful APIs
Β Flask-RESTXAPI with documentationpip install flask-restxDocumented APIs
EmailFlask-MailEmail sendingpip install flask-mailNotifications
CachingFlask-CachingCaching frameworkpip install flask-cachingPerformance
SecurityFlask-TalismanSecurity headerspip install flask-talismanHTTPS, headers
AdminFlask-AdminAdmin interfacepip install flask-adminAdmin panel
MonitoringFlask-MonitoringDashboardPerformance monitoringpip install flask-monitoringdashboardMetrics

Table 7: Configuration Environments

EnvironmentPurposeDebug ModeDatabaseSecret KeyUse Case
DevelopmentLocal codingDEBUG=TrueSQLite or local DBCan be weakActive development
TestingAutomated testsTESTING=TrueIn-memory SQLiteTest keyCI/CD pipeline
StagingPre-productionDEBUG=FalseProduction-like DBStrong keyQA testing
ProductionLive applicationDEBUG=FalseProduction DBStrong from envReal users

Complete Example Application

This is a fully functional Flask blog application demonstrating best practices and production-ready patterns.

Project Structure

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
flask_blog/
β”œβ”€β”€ app/
β”‚   β”œβ”€β”€ __init__.py
β”‚   β”œβ”€β”€ models.py
β”‚   β”œβ”€β”€ auth/
β”‚   β”‚   β”œβ”€β”€ __init__.py
β”‚   β”‚   β”œβ”€β”€ routes.py
β”‚   β”‚   └── forms.py
β”‚   β”œβ”€β”€ blog/
β”‚   β”‚   β”œβ”€β”€ __init__.py
β”‚   β”‚   β”œβ”€β”€ routes.py
β”‚   β”‚   └── forms.py
β”‚   β”œβ”€β”€ api/
β”‚   β”‚   β”œβ”€β”€ __init__.py
β”‚   β”‚   └── routes.py
β”‚   β”œβ”€β”€ templates/
β”‚   β”‚   β”œβ”€β”€ base.html
β”‚   β”‚   β”œβ”€β”€ index.html
β”‚   β”‚   β”œβ”€β”€ auth/
β”‚   β”‚   β”‚   β”œβ”€β”€ login.html
β”‚   β”‚   β”‚   └── register.html
β”‚   β”‚   └── blog/
β”‚   β”‚       β”œβ”€β”€ index.html
β”‚   β”‚       β”œβ”€β”€ post.html
β”‚   β”‚       └── create.html
β”‚   └── static/
β”‚       β”œβ”€β”€ css/
β”‚       β”‚   └── style.css
β”‚       └── js/
β”‚           └── main.js
β”œβ”€β”€ migrations/
β”œβ”€β”€ tests/
β”‚   β”œβ”€β”€ __init__.py
β”‚   β”œβ”€β”€ conftest.py
β”‚   β”œβ”€β”€ test_models.py
β”‚   β”œβ”€β”€ test_auth.py
β”‚   └── test_blog.py
β”œβ”€β”€ config.py
β”œβ”€β”€ run.py
β”œβ”€β”€ requirements.txt
β”œβ”€β”€ .env.example
└── .gitignore

1. Application Factory (app/__init__.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
from config import config

# Initialize extensions
db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
csrf = CSRFProtect()

def create_app(config_name='development'):
    """
    Application factory function.
    
    Args:
        config_name: Configuration environment (development, testing, production)
        
    Returns:
        Configured Flask application instance
    """
    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)
    csrf.init_app(app)
    
    # Configure Flask-Login
    login_manager.login_view = 'auth.login'
    login_manager.login_message = 'Please log in to access this page.'
    login_manager.login_message_category = 'info'
    
    # Register blueprints
    from app.auth import auth_bp
    from app.blog import blog_bp
    from app.api import api_bp
    
    app.register_blueprint(auth_bp, url_prefix='/auth')
    app.register_blueprint(blog_bp, url_prefix='/blog')
    app.register_blueprint(api_bp, url_prefix='/api')
    
    # Register main routes
    from app.main import main_bp
    app.register_blueprint(main_bp)
    
    # Register error handlers
    register_error_handlers(app)
    
    # Shell context for flask cli
    @app.shell_context_processor
    def make_shell_context():
        return {
            'db': db,
            'User': User,
            'Post': Post,
            'Comment': Comment
        }
    
    # Custom CLI commands
    register_commands(app)
    
    return app

def register_error_handlers(app):
    """Register custom error handlers"""
    
    @app.errorhandler(404)
    def not_found_error(error):
        if request.path.startswith('/api/'):
            return {'error': 'Resource not found'}, 404
        return render_template('errors/404.html'), 404
    
    @app.errorhandler(403)
    def forbidden_error(error):
        if request.path.startswith('/api/'):
            return {'error': 'Forbidden'}, 403
        return render_template('errors/403.html'), 403
    
    @app.errorhandler(500)
    def internal_error(error):
        db.session.rollback()
        if request.path.startswith('/api/'):
            return {'error': 'Internal server error'}, 500
        return render_template('errors/500.html'), 500

def register_commands(app):
    """Register custom CLI commands"""
    import click
    
    @app.cli.command()
    def init_db():
        """Initialize the database."""
        db.create_all()
        click.echo('Database initialized.')
    
    @app.cli.command()
    def seed_db():
        """Seed database with sample data."""
        from app.models import User, Post
        
        # Create sample users
        admin = User(username='admin', email='admin@example.com', is_admin=True)
        admin.set_password('admin123')
        
        user1 = User(username='john', email='john@example.com')
        user1.set_password('password')
        
        db.session.add_all([admin, user1])
        db.session.commit()
        
        # Create sample posts
        post1 = Post(
            title='Welcome to Flask Blog',
            body='This is the first post on our blog!',
            author=admin
        )
        post2 = Post(
            title='Getting Started with Flask',
            body='Flask is a micro web framework...',
            author=user1
        )
        
        db.session.add_all([post1, post2])
        db.session.commit()
        
        click.echo('Database seeded with sample data.')

# Import models to avoid circular imports
from app.models import User, Post, Comment
from flask import request, render_template

2. Models (app/models.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
from app import db, login_manager
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin

@login_manager.user_loader
def load_user(user_id):
    """Load user by ID for Flask-Login"""
    return User.query.get(int(user_id))

class User(UserMixin, db.Model):
    """User model for authentication and authorization"""
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False, index=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    is_active = db.Column(db.Boolean, default=True)
    is_admin = db.Column(db.Boolean, default=False)
    bio = db.Column(db.Text)
    
    # Relationships
    posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
    comments = db.relationship('Comment', backref='author', lazy='dynamic', cascade='all, delete-orphan')
    
    def set_password(self, password):
        """Hash and set password"""
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        """Verify password against hash"""
        return check_password_hash(self.password_hash, password)
    
    def to_dict(self, include_email=False):
        """Serialize user to dictionary"""
        data = {
            'id': self.id,
            'username': self.username,
            'created_at': self.created_at.isoformat(),
            'is_active': self.is_active,
            'post_count': self.posts.count()
        }
        if include_email:
            data['email'] = self.email
        return data
    
    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    """Blog post model"""
    __tablename__ = 'posts'
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    slug = db.Column(db.String(200), unique=True, nullable=False, index=True)
    body = db.Column(db.Text, nullable=False)
    summary = db.Column(db.String(500))
    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    published = db.Column(db.Boolean, default=False)
    views = db.Column(db.Integer, default=0)
    
    # Foreign key
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    
    # Relationships
    comments = db.relationship('Comment', backref='post', lazy='dynamic', cascade='all, delete-orphan')
    
    def generate_slug(self):
        """Generate URL-safe slug from title"""
        import re
        slug = re.sub(r'[^\w\s-]', '', self.title.lower())
        slug = re.sub(r'[-\s]+', '-', slug)
        
        # Ensure uniqueness
        base_slug = slug
        counter = 1
        while Post.query.filter_by(slug=slug).first():
            slug = f"{base_slug}-{counter}"
            counter += 1
        
        return slug
    
    def to_dict(self, include_body=False):
        """Serialize post to dictionary"""
        data = {
            'id': self.id,
            'title': self.title,
            'slug': self.slug,
            'summary': self.summary,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat(),
            'published': self.published,
            'views': self.views,
            'author': self.author.username,
            'comment_count': self.comments.count()
        }
        if include_body:
            data['body'] = self.body
        return data
    
    def __repr__(self):
        return f'<Post {self.title}>'

class Comment(db.Model):
    """Comment model"""
    __tablename__ = 'comments'
    
    id = db.Column(db.Integer, primary_key=True)
    body = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    approved = db.Column(db.Boolean, default=True)
    
    # Foreign keys
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    post_id = db.Column(db.Integer, db.ForeignKey('posts.id'), nullable=False)
    
    def to_dict(self):
        """Serialize comment to dictionary"""
        return {
            'id': self.id,
            'body': self.body,
            'created_at': self.created_at.isoformat(),
            'author': self.author.username,
            'post_id': self.post_id
        }
    
    def __repr__(self):
        return f'<Comment {self.id}>'

3. Authentication Blueprint (app/auth/__init__.py)

1
2
3
4
5
from flask import Blueprint

auth_bp = Blueprint('auth', __name__)

from app.auth import routes

4. Authentication Routes (app/auth/routes.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from app.auth import auth_bp
from app.auth.forms import LoginForm, RegistrationForm
from app.models import User
from app import db

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    """User registration"""
    if current_user.is_authenticated:
        return redirect(url_for('blog.index'))
    
    form = RegistrationForm()
    if form.validate_on_submit():
        # Check if user already exists
        if User.query.filter_by(email=form.email.data).first():
            flash('Email already registered', 'error')
            return redirect(url_for('auth.register'))
        
        if User.query.filter_by(username=form.username.data).first():
            flash('Username already taken', 'error')
            return redirect(url_for('auth.register'))
        
        # Create new user
        user = User(username=form.username.data, email=form.email.data)
        user.set_password(form.password.data)
        
        db.session.add(user)
        db.session.commit()
        
        flash('Registration successful! Please log in.', 'success')
        return redirect(url_for('auth.login'))
    
    return render_template('auth/register.html', form=form)

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    """User login"""
    if current_user.is_authenticated:
        return redirect(url_for('blog.index'))
    
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        
        if user and user.check_password(form.password.data):
            login_user(user, remember=form.remember.data)
            
            # Redirect to next page or home
            next_page = request.args.get('next')
            if not next_page or not next_page.startswith('/'):
                next_page = url_for('blog.index')
            
            flash(f'Welcome back, {user.username}!', 'success')
            return redirect(next_page)
        
        flash('Invalid email or password', 'error')
    
    return render_template('auth/login.html', form=form)

@auth_bp.route('/logout')
@login_required
def logout():
    """User logout"""
    logout_user()
    flash('You have been logged out', 'info')
    return redirect(url_for('main.index'))

@auth_bp.route('/profile')
@login_required
def profile():
    """User profile"""
    return render_template('auth/profile.html', user=current_user)

5. Authentication Forms (app/auth/forms.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Email, EqualTo, Length, ValidationError
from app.models import User

class LoginForm(FlaskForm):
    """User login form"""
    email = StringField('Email', validators=[
        DataRequired(),
        Email()
    ])
    password = PasswordField('Password', validators=[DataRequired()])
    remember = BooleanField('Remember Me')
    submit = SubmitField('Log In')

class RegistrationForm(FlaskForm):
    """User registration form"""
    username = StringField('Username', validators=[
        DataRequired(),
        Length(min=3, max=80)
    ])
    email = StringField('Email', validators=[
        DataRequired(),
        Email(),
        Length(max=120)
    ])
    password = PasswordField('Password', validators=[
        DataRequired(),
        Length(min=6)
    ])
    password_confirm = PasswordField('Confirm Password', validators=[
        DataRequired(),
        EqualTo('password', message='Passwords must match')
    ])
    submit = SubmitField('Register')
    
    def validate_username(self, field):
        """Check if username is already taken"""
        if User.query.filter_by(username=field.data).first():
            raise ValidationError('Username already taken')
    
    def validate_email(self, field):
        """Check if email is already registered"""
        if User.query.filter_by(email=field.data).first():
            raise ValidationError('Email already registered')

6. Blog Blueprint (app/blog/__init__.py)

1
2
3
4
5
from flask import Blueprint

blog_bp = Blueprint('blog', __name__)

from app.blog import routes

7. Blog Routes (app/blog/routes.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
from flask import render_template, redirect, url_for, flash, request, abort
from flask_login import login_required, current_user
from app.blog import blog_bp
from app.blog.forms import PostForm, CommentForm
from app.models import Post, Comment
from app import db

@blog_bp.route('/')
def index():
    """List all published posts"""
    page = request.args.get('page', 1, type=int)
    per_page = 10
    
    query = Post.query.filter_by(published=True).order_by(Post.created_at.desc())
    pagination = query.paginate(page=page, per_page=per_page, error_out=False)
    
    return render_template('blog/index.html', posts=pagination.items, pagination=pagination)

@blog_bp.route('/post/<slug>')
def post(slug):
    """View single post"""
    post = Post.query.filter_by(slug=slug, published=True).first_or_404()
    
    # Increment view count
    post.views += 1
    db.session.commit()
    
    # Get comments
    comments = post.comments.filter_by(approved=True).order_by(Comment.created_at.desc()).all()
    
    comment_form = CommentForm()
    
    return render_template('blog/post.html', post=post, comments=comments, form=comment_form)

@blog_bp.route('/create', methods=['GET', 'POST'])
@login_required
def create():
    """Create new post"""
    form = PostForm()
    
    if form.validate_on_submit():
        post = Post(
            title=form.title.data,
            body=form.body.data,
            summary=form.summary.data,
            published=form.published.data,
            author=current_user
        )
        post.slug = post.generate_slug()
        
        db.session.add(post)
        db.session.commit()
        
        flash('Post created successfully!', 'success')
        return redirect(url_for('blog.post', slug=post.slug))
    
    return render_template('blog/create.html', form=form)

@blog_bp.route('/edit/<int:post_id>', methods=['GET', 'POST'])
@login_required
def edit(post_id):
    """Edit existing post"""
    post = Post.query.get_or_404(post_id)
    
    # Check authorization
    if post.author != current_user and not current_user.is_admin:
        abort(403)
    
    form = PostForm(obj=post)
    
    if form.validate_on_submit():
        post.title = form.title.data
        post.body = form.body.data
        post.summary = form.summary.data
        post.published = form.published.data
        
        # Regenerate slug if title changed
        if post.title != form.title.data:
            post.slug = post.generate_slug()
        
        db.session.commit()
        
        flash('Post updated successfully!', 'success')
        return redirect(url_for('blog.post', slug=post.slug))
    
    return render_template('blog/edit.html', form=form, post=post)

@blog_bp.route('/delete/<int:post_id>', methods=['POST'])
@login_required
def delete(post_id):
    """Delete post"""
    post = Post.query.get_or_404(post_id)
    
    # Check authorization
    if post.author != current_user and not current_user.is_admin:
        abort(403)
    
    db.session.delete(post)
    db.session.commit()
    
    flash('Post deleted successfully!', 'success')
    return redirect(url_for('blog.index'))

@blog_bp.route('/post/<slug>/comment', methods=['POST'])
@login_required
def add_comment(slug):
    """Add comment to post"""
    post = Post.query.filter_by(slug=slug).first_or_404()
    form = CommentForm()
    
    if form.validate_on_submit():
        comment = Comment(
            body=form.body.data,
            author=current_user,
            post=post
        )
        
        db.session.add(comment)
        db.session.commit()
        
        flash('Comment added successfully!', 'success')
    
    return redirect(url_for('blog.post', slug=slug))

@blog_bp.route('/my-posts')
@login_required
def my_posts():
    """View user's own posts"""
    page = request.args.get('page', 1, type=int)
    pagination = current_user.posts.order_by(Post.created_at.desc()).paginate(
        page=page, per_page=10, error_out=False
    )
    
    return render_template('blog/my_posts.html', posts=pagination.items, pagination=pagination)

8. Blog Forms (app/blog/forms.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from flask_wtf import FlaskForm
from wtforms import StringField, TextAreaField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length

class PostForm(FlaskForm):
    """Blog post form"""
    title = StringField('Title', validators=[
        DataRequired(),
        Length(min=5, max=200)
    ])
    summary = StringField('Summary', validators=[
        Length(max=500)
    ])
    body = TextAreaField('Content', validators=[
        DataRequired(),
        Length(min=20)
    ])
    published = BooleanField('Publish immediately')
    submit = SubmitField('Save Post')

class CommentForm(FlaskForm):
    """Comment form"""
    body = TextAreaField('Comment', validators=[
        DataRequired(),
        Length(min=5, max=1000)
    ])
    submit = SubmitField('Add Comment')

9. API Blueprint (app/api/__init__.py)

1
2
3
4
5
from flask import Blueprint

api_bp = Blueprint('api', __name__)

from app.api import routes

10. API Routes (app/api/routes.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from flask import jsonify, request
from app.api import api_bp
from app.models import User, Post, Comment
from app import db

@api_bp.route('/posts', methods=['GET'])
def get_posts():
    """Get all published posts"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    query = Post.query.filter_by(published=True).order_by(Post.created_at.desc())
    pagination = query.paginate(page=page, per_page=min(per_page, 100), error_out=False)
    
    return jsonify({
        'posts': [post.to_dict() for post in pagination.items],
        'total': pagination.total,
        'page': page,
        'per_page': per_page,
        'pages': pagination.pages
    })

@api_bp.route('/posts/<slug>', methods=['GET'])
def get_post(slug):
    """Get single post by slug"""
    post = Post.query.filter_by(slug=slug, published=True).first_or_404()
    return jsonify(post.to_dict(include_body=True))

@api_bp.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    """Get user by ID"""
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

@api_bp.route('/posts/<slug>/comments', methods=['GET'])
def get_comments(slug):
    """Get comments for a post"""
    post = Post.query.filter_by(slug=slug).first_or_404()
    comments = post.comments.filter_by(approved=True).order_by(Comment.created_at.desc()).all()
    
    return jsonify({
        'comments': [comment.to_dict() for comment in comments],
        'count': len(comments)
    })

11. Main Blueprint (app/main/__init__.py)

1
2
3
4
5
from flask import Blueprint

main_bp = Blueprint('main', __name__)

from app.main import routes

12. Main Routes (app/main/routes.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from flask import render_template
from app.main import main_bp
from app.models import Post

@main_bp.route('/')
def index():
    """Home page"""
    recent_posts = Post.query.filter_by(published=True).order_by(
        Post.created_at.desc()
    ).limit(5).all()
    
    return render_template('index.html', posts=recent_posts)

@main_bp.route('/about')
def about():
    """About page"""
    return render_template('about.html')

13. Configuration (config.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import os
from datetime import timedelta

basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    """Base configuration"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_RECORD_QUERIES = True
    
    # Session configuration
    PERMANENT_SESSION_LIFETIME = timedelta(days=7)
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Lax'
    
    # WTF Forms
    WTF_CSRF_ENABLED = True
    WTF_CSRF_TIME_LIMIT = None
    
    # Pagination
    POSTS_PER_PAGE = 10
    COMMENTS_PER_PAGE = 20
    
    @staticmethod
    def init_app(app):
        """Initialize application"""
        pass

class DevelopmentConfig(Config):
    """Development configuration"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'dev-db.sqlite')
    SQLALCHEMY_ECHO = True
    SESSION_COOKIE_SECURE = False  # Allow HTTP in development

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

class ProductionConfig(Config):
    """Production configuration"""
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')
    
    @classmethod
    def init_app(cls, app):
        Config.init_app(app)
        
        # Log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

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

14. Application Entry Point (run.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import os
from app import create_app, db
from app.models import User, Post, Comment

# Determine configuration from environment
config_name = os.getenv('FLASK_CONFIG', 'development')
app = create_app(config_name)

@app.shell_context_processor
def make_shell_context():
    """Make database models available in shell"""
    return {
        'db': db,
        'User': User,
        'Post': Post,
        'Comment': Comment
    }

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

15. Requirements (requirements.txt)

1
2
3
4
5
6
7
8
9
Flask==3.0.0
Flask-SQLAlchemy==3.1.1
Flask-Migrate==4.0.5
Flask-Login==0.6.3
Flask-WTF==1.2.1
WTForms==3.1.1
email-validator==2.1.0
python-dotenv==1.0.0
gunicorn==21.2.0

16. Environment Variables (.env.example)

1
2
3
4
5
6
7
8
9
10
11
12
# Flask Configuration
FLASK_APP=run.py
FLASK_ENV=development
FLASK_CONFIG=development
SECRET_KEY=your-secret-key-here

# Database
DATABASE_URL=sqlite:///app.db
DEV_DATABASE_URL=sqlite:///dev-db.sqlite

# Security
WTF_CSRF_SECRET_KEY=your-csrf-secret-key

17. Git Ignore (.gitignore)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual Environment
venv/
env/
ENV/

# Flask
instance/
.webassets-cache

# Database
*.sqlite
*.db

# Environment
.env
.flaskenv

# IDE
.vscode/
.idea/
*.swp
*.swo

# Logs
*.log

# OS
.DS_Store
Thumbs.db

18. Running the Application

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Set up environment variables
cp .env.example .env
# Edit .env with your configuration

# Initialize database
flask db init
flask db migrate -m "Initial migration"
flask db upgrade

# Seed database (optional)
flask seed-db

# Run development server
flask run

# Or using run.py
python run.py

# Access application at http://localhost:5000

Key Features of This Example:

βœ… Application Factory Pattern - Proper app initialization
βœ… Blueprint Organization - Modular structure
βœ… Database Models - User, Post, Comment with relationships
βœ… Authentication System - Registration, login, logout
βœ… Authorization - Role-based access control
βœ… CRUD Operations - Create, read, update, delete posts
βœ… Comment System - Nested comments on posts
βœ… Form Validation - WTForms with validators
βœ… API Endpoints - RESTful API with pagination
βœ… Error Handling - Custom error pages
βœ… CLI Commands - Database initialization and seeding
βœ… Multiple Configurations -

This post is licensed under CC BY 4.0 by the author.