Table of Contents
- Introduction
- Core Concepts
- Application Structure
- Application Factory Pattern
- Blueprints
- Configuration Management
- Database Integration
- Request Handling
- Templates and Static Files
- REST API Development
- Authentication and Authorization
- Security Best Practices
- Testing
- Deployment
- Performance Optimization
- Best Practices
- Common Pitfalls
- 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:
- Client Request: Browser/client sends HTTP request
- URL Routing: Flask matches URL to a view function
- View Processing: Function processes request and generates response
- 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
- Multiple Instances: Create apps with different configurations
- Testing: Easily create test instances with test configurations
- Avoid Circular Imports: Extensions initialized before routes
- Configuration Flexibility: Dynamic configuration based on environment
- 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
- Start Simple: Flaskβs micro-framework nature makes it perfect for beginners and prototyping
- Scale Thoughtfully: Use application factory and blueprints for larger applications
- Security First: Always implement CSRF protection, secure passwords, and validate input
- Test Everything: Write comprehensive tests for models, routes, and APIs
- Monitor Performance: Use caching, optimize queries, and implement proper logging
- 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>© 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>© {{ 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'})
|
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
|
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()
|
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
|
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')
|
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)
|
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
|
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
|
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 Term | Alternative Terms | Definition | Context |
|---|
| Route | endpoint, URL pattern, view mapping | URL pattern mapped to a view function | URL routing |
| View Function | route handler, controller, endpoint function | Function that handles HTTP requests | Request handling |
| Blueprint | module, sub-application, component | Modular component of Flask application | Application structure |
| Application Factory | app factory, factory pattern | Function that creates Flask application instances | Application creation |
| Context | request context, application context | Environment data available during request | Request lifecycle |
| Extension | plugin, add-on, library integration | Third-party library integrated with Flask | Feature enhancement |
| Template | view, HTML template, Jinja template | HTML file with dynamic content placeholders | Frontend rendering |
| Static Files | assets, resources, media files | CSS, JavaScript, images, fonts | Frontend resources |
| Migration | schema change, database evolution | Database schema version control | Database management |
| WSGI | Web Server Gateway Interface | Standard interface between web server and app | Deployment |
| Request Object | req, HTTP request | Object containing request data | Request handling |
| Response Object | HTTP response, reply | Object returned to client | Response handling |
| Session | user session, cookie session | User-specific data storage | State management |
| Flash Message | alert, notification, temporary message | One-time message displayed to user | User feedback |
| g Object | global, request-local storage | Request-local data storage | Data sharing |
Table 2: Hierarchical Application Structure
| Level | Component | Sub-Component | Purpose | File/Folder |
|---|
| 1 | Application Root | Β | Top-level application | / |
| 2 | Application Package | Β | Main application code | /app/ |
| Β | Β | Models | Database models | /app/models.py |
| Β | Β | Views/Routes | Request handlers | /app/routes.py |
| Β | Β | Forms | Form definitions | /app/forms.py |
| Β | Β | Blueprints | Modular components | /app/auth/, /app/blog/ |
| 2 | Configuration | Β | Application settings | /config.py |
| Β | Β | Development Config | Dev environment settings | Class in config.py |
| Β | Β | Production Config | Prod environment settings | Class in config.py |
| Β | Β | Testing Config | Test environment settings | Class in config.py |
| 2 | Templates | Β | HTML templates | /app/templates/ |
| Β | Β | Base Templates | Layout templates | /app/templates/base.html |
| Β | Β | Blueprint Templates | Component-specific templates | /app/templates/auth/ |
| 2 | Static Files | Β | Frontend assets | /app/static/ |
| Β | Β | CSS | Stylesheets | /app/static/css/ |
| Β | Β | JavaScript | Client-side scripts | /app/static/js/ |
| Β | Β | Images | Graphics and icons | /app/static/images/ |
| 2 | Migrations | Β | Database schema versions | /migrations/ |
| 2 | Tests | Β | Test suite | /tests/ |
| Β | Β | Unit Tests | Component tests | /tests/test_models.py |
| Β | Β | Integration Tests | System tests | /tests/test_routes.py |
| 2 | Instance Folder | Β | Instance-specific files | /instance/ |
Table 3: Request-Response Cycle Stages
| Stage | Technical Term | Description | Flask Components | HTTP Layer |
|---|
| 1 | Request Initiation | Client sends HTTP request | WSGI server receives request | HTTP Request |
| 2 | Request Parsing | Parse HTTP request data | Werkzeug processes request | Headers, Body, Method |
| 3 | URL Routing | Match URL to view function | Flask router, @app.route() | URL, Path Parameters |
| 4 | Before Request | Pre-processing hooks | @app.before_request | N/A |
| 5 | Context Creation | Push request/app contexts | Request context, App context | N/A |
| 6 | View Execution | Execute matched view function | View function, Business logic | N/A |
| 7 | Response Creation | Generate HTTP response | make_response(), jsonify() | Status Code, Headers |
| 8 | After Request | Post-processing hooks | @app.after_request | N/A |
| 9 | Context Teardown | Clean up contexts | @app.teardown_request | N/A |
| 10 | Response Sending | Send response to client | WSGI server, HTTP | HTTP Response |
Table 4: Database Operations Terminology
| Operation | SQL Equivalent | SQLAlchemy Method | Flask-SQLAlchemy | Use Case |
|---|
| Create | INSERT | session.add() | db.session.add() | Add new record |
| Read | SELECT | query.filter() | Model.query.filter_by() | Retrieve records |
| Update | UPDATE | Modify object + commit | Modify + db.session.commit() | Change existing record |
| Delete | DELETE | session.delete() | db.session.delete() | Remove record |
| Query | SELECT | query() | Model.query | Build query |
| Filter | WHERE | filter(), filter_by() | .filter_by(field=value) | Conditional selection |
| Join | JOIN | join() | .join(Model) | Combine tables |
| Order | ORDER BY | order_by() | .order_by(Model.field) | Sort results |
| Limit | LIMIT | limit() | .limit(n) | Restrict count |
| Count | COUNT(*) | count() | .count() | Count records |
| Pagination | LIMIT + OFFSET | paginate() | .paginate(page, per_page) | Paginated results |
| Commit | COMMIT | session.commit() | db.session.commit() | Save changes |
| Rollback | ROLLBACK | session.rollback() | db.session.rollback() | Undo changes |
Table 5: HTTP Methods and Flask Decorators
| HTTP Method | Purpose | Idempotent | Safe | Flask Decorator | Typical Use |
|---|
| GET | Retrieve resource | Yes | Yes | @app.route('/', methods=['GET']) | View data, listings |
| POST | Create resource | No | No | @app.route('/', methods=['POST']) | Submit forms, create |
| PUT | Update/Replace resource | Yes | No | @app.route('/<id>', methods=['PUT']) | Full update |
| PATCH | Partial update | No | No | @app.route('/<id>', methods=['PATCH']) | Partial update |
| DELETE | Delete resource | Yes | No | @app.route('/<id>', methods=['DELETE']) | Remove resource |
| HEAD | Get headers only | Yes | Yes | Automatic with GET | Check existence |
| OPTIONS | Get allowed methods | Yes | Yes | Automatic | CORS preflight |
Table 6: Extension Categories
| Category | Extension Name | Purpose | Installation | Common Use |
|---|
| Database | Flask-SQLAlchemy | ORM for databases | pip install flask-sqlalchemy | Database models |
| Β | Flask-Migrate | Database migrations | pip install flask-migrate | Schema versioning |
| Authentication | Flask-Login | User session management | pip install flask-login | Login/logout |
| Β | Flask-JWT-Extended | JWT authentication | pip install flask-jwt-extended | API authentication |
| Forms | Flask-WTF | Form handling and CSRF | pip install flask-wtf | Form validation |
| API | Flask-RESTful | REST API framework | pip install flask-restful | RESTful APIs |
| Β | Flask-RESTX | API with documentation | pip install flask-restx | Documented APIs |
| Email | Flask-Mail | Email sending | pip install flask-mail | Notifications |
| Caching | Flask-Caching | Caching framework | pip install flask-caching | Performance |
| Security | Flask-Talisman | Security headers | pip install flask-talisman | HTTPS, headers |
| Admin | Flask-Admin | Admin interface | pip install flask-admin | Admin panel |
| Monitoring | Flask-MonitoringDashboard | Performance monitoring | pip install flask-monitoringdashboard | Metrics |
Table 7: Configuration Environments
| Environment | Purpose | Debug Mode | Database | Secret Key | Use Case |
|---|
| Development | Local coding | DEBUG=True | SQLite or local DB | Can be weak | Active development |
| Testing | Automated tests | TESTING=True | In-memory SQLite | Test key | CI/CD pipeline |
| Staging | Pre-production | DEBUG=False | Production-like DB | Strong key | QA testing |
| Production | Live application | DEBUG=False | Production DB | Strong from env | Real 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)
|
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 -