Table of Contents
Quick Navigation
- Flask Architecture
- Application Factory Pattern
- Blueprints & Modular Design
- Configuration Management
- Database Integration
- Authentication & Authorization
- REST API Development
- Flask Extensions
- Error Handling
- Testing Strategies
- Security Best Practices
- Performance Optimization
- Production Deployment
- Monitoring & Logging
- Resources & Learning Path
Flask Architecture
Request/Response Cycle
┌─────────────────────────────────────────────────────────┐
│ Client Request │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ WSGI Server │
│ (Gunicorn, uWSGI, Waitress) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Flask Application │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Application Context │ │
│ │ Request Context │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Before Request Hooks │
│ ┌──────────────────────────────────────────────────┐ │
│ │ @app.before_request │ │
│ │ @app.before_first_request │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ URL Routing │
│ (Werkzeug routing system) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ View Function │
│ ┌──────────────────────────────────────────────────┐ │
│ │ • Process request │ │
│ │ • Query database │ │
│ │ • Business logic │ │
│ │ • Generate response │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ After Request Hooks │
│ ┌──────────────────────────────────────────────────┐ │
│ │ @app.after_request │ │
│ │ @app.teardown_request │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Template Rendering │
│ (Jinja2, if rendering templates) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Client Response │
└─────────────────────────────────────────────────────────┘
Project Structure (Production)
myproject/ ├── app/ │ ├── __init__.py # Application factory │ ├── config.py # Configuration │ │ │ ├── api/ # API blueprints │ │ ├── __init__.py │ │ ├── auth.py │ │ ├── users.py │ │ └── items.py │ │ │ ├── main/ # Main blueprint │ │ ├── __init__.py │ │ ├── routes.py │ │ └── errors.py │ │ │ ├── models/ # Database models │ │ ├── __init__.py │ │ ├── user.py │ │ └── item.py │ │ │ ├── schemas/ # Marshmallow schemas │ │ ├── __init__.py │ │ ├── user.py │ │ └── item.py │ │ │ ├── services/ # Business logic │ │ ├── __init__.py │ │ ├── user_service.py │ │ └── email_service.py │ │ │ ├── utils/ # Utilities │ │ ├── __init__.py │ │ ├── decorators.py │ │ └── validators.py │ │ │ ├── static/ # Static files │ │ ├── css/ │ │ ├── js/ │ │ └── images/ │ │ │ ├── templates/ # Jinja2 templates │ │ ├── base.html │ │ └── index.html │ │ │ └── extensions.py # Flask extensions │ ├── migrations/ # Database migrations │ └── versions/ │ ├── tests/ # Tests │ ├── __init__.py │ ├── conftest.py │ ├── test_api/ │ ├── test_models/ │ └── test_services/ │ ├── instance/ # Instance folder (not in VCS) │ └── config.py │ ├── requirements/ │ ├── base.txt │ ├── development.txt │ ├── production.txt │ └── testing.txt │ ├── docker/ │ ├── Dockerfile │ ├── docker-compose.yml │ └── nginx.conf │ ├── scripts/ │ ├── init_db.py │ └── seed_data.py │ ├── .env.example ├── .flaskenv ├── wsgi.py # WSGI entry point ├── manage.py # CLI commands ├── pytest.ini └── README.md Environment Variables (.env) FLASK_APP=wsgi.py FLASK_ENV=production SECRET_KEY=your-secret-key-here DATABASE_URL=postgresql://user:pass@localhost:5432/db REDIS_URL=redis://localhost:6379/0 MAIL_SERVER=smtp.gmail.com MAIL_PORT=587 [email protected] MAIL_PASSWORD=your-password JWT_SECRET_KEY=your-jwt-secret
Basic Flask Application
# Simple Flask app (not for production)
from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/')
def index():
return jsonify({"message": "Hello, World!"})
@app.route('/users/')
def get_user(user_id):
return jsonify({"user_id": user_id})
if __name__ == '__main__':
app.run(debug=True)
# Flask with configuration
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
# Request handling
from flask import request, jsonify
@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
if request.method == 'POST':
data = request.get_json()
return jsonify({"received": data}), 201
else:
return jsonify({"data": "sample"})
# Query parameters
@app.route('/search')
def search():
query = request.args.get('q', '')
page = request.args.get('page', 1, type=int)
return jsonify({"query": query, "page": page})
# Path variables
@app.route('/posts/')
def get_post(post_id):
return jsonify({"post_id": post_id})
@app.route('/users/')
def get_user_by_name(username):
return jsonify({"username": username})
# JSON response
@app.route('/api/users')
def get_users():
users = [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"}
]
return jsonify(users)
# Response with status code
@app.route('/api/create', methods=['POST'])
def create():
data = request.get_json()
# Create resource
return jsonify({"id": 123}), 201
# Error responses
from flask import abort
@app.route('/api/items/')
def get_item(item_id):
item = find_item(item_id)
if not item:
abort(404)
return jsonify(item)
# Redirects
from flask import redirect, url_for
@app.route('/old-path')
def old_path():
return redirect(url_for('new_path'))
@app.route('/new-path')
def new_path():
return jsonify({"message": "New path"})
# Cookies
from flask import make_response
@app.route('/set-cookie')
def set_cookie():
response = make_response(jsonify({"message": "Cookie set"}))
response.set_cookie('user_id', '123', max_age=3600)
return response
@app.route('/get-cookie')
def get_cookie():
user_id = request.cookies.get('user_id')
return jsonify({"user_id": user_id})
# Sessions
from flask import session
app.config['SECRET_KEY'] = 'your-secret-key'
@app.route('/login', methods=['POST'])
def login():
session['user_id'] = 123
return jsonify({"message": "Logged in"})
@app.route('/logout')
def logout():
session.pop('user_id', None)
return jsonify({"message": "Logged out"})
@app.route('/profile')
def profile():
user_id = session.get('user_id')
if not user_id:
abort(401)
return jsonify({"user_id": user_id})
Application Factory Pattern
Application Factory
# app/__init__.py
from flask import Flask
from app.extensions import db, migrate, login_manager, cors, ma
from app.config import config
def create_app(config_name='default'):
"""Application factory pattern"""
app = Flask(__name__)
# Load configuration
app.config.from_object(config[config_name])
config[config_name].init_app(app)
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
login_manager.init_app(app)
cors.init_app(app)
ma.init_app(app)
# Register blueprints
from app.api import api_bp
app.register_blueprint(api_bp, url_prefix='/api')
from app.main import main_bp
app.register_blueprint(main_bp)
# Register error handlers
register_error_handlers(app)
# Register CLI commands
register_cli_commands(app)
# Setup logging
setup_logging(app)
return app
def register_error_handlers(app):
"""Register error handlers"""
from flask import jsonify
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "Not found"}), 404
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return jsonify({"error": "Internal server error"}), 500
def register_cli_commands(app):
"""Register custom CLI commands"""
@app.cli.command()
def init_db():
"""Initialize the database"""
db.create_all()
print("Database initialized")
@app.cli.command()
def seed_db():
"""Seed the database with sample data"""
# Add seed data
print("Database seeded")
def setup_logging(app):
"""Setup logging"""
if not app.debug and not app.testing:
import logging
from logging.handlers import RotatingFileHandler
import os
if not os.path.exists('logs'):
os.mkdir('logs')
file_handler = RotatingFileHandler(
'logs/app.log',
maxBytes=10240000,
backupCount=10
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s '
'[in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Application startup')
# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_cors import CORS
from flask_marshmallow import Marshmallow
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
cors = CORS()
ma = Marshmallow()
cache = Cache()
limiter = Limiter(key_func=get_remote_address)
# wsgi.py (entry point)
import os
from app import create_app
app = create_app(os.getenv('FLASK_ENV', 'production'))
if __name__ == '__main__':
app.run()
# Usage in testing
def test_app():
app = create_app('testing')
with app.app_context():
db.create_all()
# Run tests
db.drop_all()
Configuration Classes
# app/config.py
import os
from datetime import timedelta
class Config:
"""Base configuration"""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_RECORD_QUERIES = True
# Session
PERMANENT_SESSION_LIFETIME = timedelta(hours=24)
# Mail
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() == 'true'
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
# Pagination
ITEMS_PER_PAGE = 20
# File uploads
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER = os.path.join(os.getcwd(), 'uploads')
# Cache
CACHE_TYPE = 'redis'
CACHE_REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
CACHE_DEFAULT_TIMEOUT = 300
# Rate limiting
RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/1')
# JWT
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
@staticmethod
def init_app(app):
pass
class DevelopmentConfig(Config):
"""Development configuration"""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///dev.db'
SQLALCHEMY_ECHO = True
class TestingConfig(Config):
"""Testing configuration"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
"""Production configuration"""
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
# Security
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
REMEMBER_COOKIE_SECURE = True
REMEMBER_COOKIE_HTTPONLY = True
@classmethod
def init_app(cls, app):
Config.init_app(app)
# Log to syslog
import logging
from logging.handlers import SysLogHandler
syslog_handler = SysLogHandler()
syslog_handler.setLevel(logging.WARNING)
app.logger.addHandler(syslog_handler)
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
# Loading config in app
app = Flask(__name__)
app.config.from_object(config['production'])
# Or from environment variable
config_name = os.getenv('FLASK_ENV', 'production')
app.config.from_object(config[config_name])
# Load from instance folder
app.config.from_pyfile('config.py', silent=True)
# Load from environment variables
app.config.from_prefixed_env()
Blueprints & Modular Design
Blueprint Basics
# app/api/__init__.py
from flask import Blueprint
api_bp = Blueprint('api', __name__)
from app.api import users, items, auth
# app/api/users.py
from flask import jsonify, request
from app.api import api_bp
from app.models import User
from app.extensions import db
@api_bp.route('/users', methods=['GET'])
def get_users():
users = User.query.all()
return jsonify([user.to_dict() for user in users])
@api_bp.route('/users/', methods=['GET'])
def get_user(user_id):
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
@api_bp.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
user = User(
email=data['email'],
name=data['name']
)
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@api_bp.route('/users/', methods=['PUT'])
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()
user.name = data.get('name', user.name)
user.email = data.get('email', user.email)
db.session.commit()
return jsonify(user.to_dict())
@api_bp.route('/users/', methods=['DELETE'])
def delete_user(user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# Register blueprint in app
from app.api import api_bp
app.register_blueprint(api_bp, url_prefix='/api')
# Multiple blueprints
# Admin blueprint
from flask import Blueprint
admin_bp = Blueprint('admin', __name__)
@admin_bp.route('/dashboard')
def dashboard():
return jsonify({"page": "admin dashboard"})
# Register with different prefix
app.register_blueprint(admin_bp, url_prefix='/admin')
# Blueprint with subdomain
api_bp = Blueprint('api', __name__, subdomain='api')
app.register_blueprint(api_bp)
# URL: api.example.com/users
# Blueprint with template folder
main_bp = Blueprint(
'main',
__name__,
template_folder='templates',
static_folder='static'
)
# Blueprint error handlers
@api_bp.errorhandler(404)
def api_not_found(error):
return jsonify({"error": "API endpoint not found"}), 404
@api_bp.errorhandler(ValidationError)
def validation_error(error):
return jsonify({"error": str(error)}), 400
# Blueprint before/after request
@api_bp.before_request
def before_api_request():
# Verify API key
api_key = request.headers.get('X-API-Key')
if not verify_api_key(api_key):
abort(401)
@api_bp.after_request
def after_api_request(response):
response.headers['X-API-Version'] = '1.0'
return response
Advanced Blueprint Patterns
# Nested blueprints (Flask 2.0+)
# app/api/v1/__init__.py
from flask import Blueprint
api_v1_bp = Blueprint('api_v1', __name__)
from app.api.v1 import users, items
# app/api/v2/__init__.py
api_v2_bp = Blueprint('api_v2', __name__)
from app.api.v2 import users, items
# Register with different prefixes
app.register_blueprint(api_v1_bp, url_prefix='/api/v1')
app.register_blueprint(api_v2_bp, url_prefix='/api/v2')
# Blueprint factory
def create_api_blueprint(name, version):
bp = Blueprint(f'api_{name}', __name__)
@bp.route('/info')
def info():
return jsonify({
"name": name,
"version": version
})
return bp
# Usage
users_v1 = create_api_blueprint('users', 'v1')
users_v2 = create_api_blueprint('users', 'v2')
# Resource-based blueprint pattern
class UserAPI:
"""RESTful user resource"""
def __init__(self, blueprint):
self.bp = blueprint
self.register_routes()
def register_routes(self):
self.bp.add_url_rule(
'/users',
'list_users',
self.list_users,
methods=['GET']
)
self.bp.add_url_rule(
'/users',
'create_user',
self.create_user,
methods=['POST']
)
self.bp.add_url_rule(
'/users/',
'get_user',
self.get_user,
methods=['GET']
)
def list_users(self):
users = User.query.all()
return jsonify([u.to_dict() for u in users])
def create_user(self):
data = request.get_json()
user = User(**data)
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
def get_user(self, user_id):
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
# Usage
api_bp = Blueprint('api', __name__)
UserAPI(api_bp)
# Blueprint with context processor
@main_bp.context_processor
def inject_user():
return dict(current_user=get_current_user())
# Blueprint with URL value preprocessor
@api_bp.url_value_preprocessor
def get_api_version(endpoint, values):
values['api_version'] = request.headers.get('API-Version', '1.0')
@api_bp.route('/data')
def get_data(api_version):
return jsonify({"api_version": api_version})
# Blueprint with template filter
@main_bp.app_template_filter('format_currency')
def format_currency(value):
return f"${value:,.2f}"
# Modular app structure with blueprints
# app/modules/users/routes.py
from flask import Blueprint
users_bp = Blueprint('users', __name__)
@users_bp.route('/')
def list_users():
pass
@users_bp.route('/')
def get_user(user_id):
pass
# app/modules/users/__init__.py
from app.modules.users.routes import users_bp
# Register all module blueprints
from app.modules.users import users_bp
from app.modules.items import items_bp
from app.modules.auth import auth_bp
app.register_blueprint(users_bp, url_prefix='/users')
app.register_blueprint(items_bp, url_prefix='/items')
app.register_blueprint(auth_bp, url_prefix='/auth')
Configuration Management
Configuration Best Practices
# Using python-dotenv
# .env file
DATABASE_URL=postgresql://localhost/mydb
SECRET_KEY=my-secret-key
DEBUG=True
# Load in app
from dotenv import load_dotenv
import os
load_dotenv()
app.config['DATABASE_URL'] = os.getenv('DATABASE_URL')
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY')
app.config['DEBUG'] = os.getenv('DEBUG', 'False') == 'True'
# Environment-specific configs
import os
class Config:
"""Base config"""
SECRET_KEY = os.environ.get('SECRET_KEY')
SQLALCHEMY_TRACK_MODIFICATIONS = False
@staticmethod
def init_app(app):
pass
class DevelopmentConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL')
SQLALCHEMY_ECHO = True
class ProductionConfig(Config):
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
@classmethod
def init_app(cls, app):
Config.init_app(app)
# Email errors to admins
import logging
from logging.handlers import SMTPHandler
credentials = None
secure = None
if getattr(cls, 'MAIL_USERNAME', None):
credentials = (cls.MAIL_USERNAME, cls.MAIL_PASSWORD)
if getattr(cls, 'MAIL_USE_TLS', None):
secure = ()
mail_handler = SMTPHandler(
mailhost=(cls.MAIL_SERVER, cls.MAIL_PORT),
fromaddr=cls.MAIL_SENDER,
toaddrs=[cls.ADMIN_EMAIL],
subject='Application Error',
credentials=credentials,
secure=secure
)
mail_handler.setLevel(logging.ERROR)
app.logger.addHandler(mail_handler)
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
# Configuration from file
# config.py
DEBUG = False
SECRET_KEY = 'secret'
DATABASE_URI = 'sqlite:///app.db'
# Load in app
app.config.from_pyfile('config.py')
# Configuration from object
class Config:
DEBUG = False
TESTING = False
DATABASE_URI = 'sqlite:///app.db'
app.config.from_object(Config)
# Configuration from JSON
# config.json
{
"DEBUG": false,
"SECRET_KEY": "secret",
"DATABASE_URI": "sqlite:///app.db"
}
# Load
import json
with open('config.json') as f:
config = json.load(f)
app.config.update(config)
# Instance folder configuration
# instance/config.py (not in version control)
SECRET_KEY = 'production-secret-key'
DATABASE_URI = 'postgresql://production-db'
# Load
app.config.from_pyfile('config.py', silent=True)
# Accessing configuration
from flask import current_app
def some_function():
secret_key = current_app.config['SECRET_KEY']
debug_mode = current_app.config.get('DEBUG', False)
# Custom configuration loader
class ConfigLoader:
@staticmethod
def load_config(app, config_name):
# Load base config
app.config.from_object(config[config_name])
# Load instance config
app.config.from_pyfile('config.py', silent=True)
# Load environment variables
app.config.update({
key[5:]: value
for key, value in os.environ.items()
if key.startswith('FLASK_')
})
# Validate required config
required = ['SECRET_KEY', 'DATABASE_URL']
for key in required:
if not app.config.get(key):
raise ValueError(f"Missing required config: {key}")
Database Integration
SQLAlchemy Models
# app/models/user.py
from app.extensions import db
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
name = db.Column(db.String(100))
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
posts = db.relationship('Post', backref='author', lazy='dynamic')
profile = db.relationship('Profile', backref='user', uselist=False)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def to_dict(self):
return {
'id': self.id,
'email': self.email,
'username': self.username,
'name': self.name,
'created_at': self.created_at.isoformat()
}
def __repr__(self):
return f''
# One-to-Many relationship
class Post(db.Model):
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# Cascade delete
comments = db.relationship(
'Comment',
backref='post',
lazy='dynamic',
cascade='all, delete-orphan'
)
# Many-to-Many relationship
# Association table
tags = db.Table('post_tags',
db.Column('post_id', db.Integer, db.ForeignKey('posts.id'), primary_key=True),
db.Column('tag_id', db.Integer, db.ForeignKey('tags.id'), primary_key=True)
)
class Post(db.Model):
# ... other fields
tags = db.relationship(
'Tag',
secondary=tags,
lazy='subquery',
backref=db.backref('posts', lazy=True)
)
class Tag(db.Model):
__tablename__ = 'tags'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
# Query patterns
# Basic queries
user = User.query.get(1) # By primary key
user = User.query.get_or_404(1) # Or 404 error
users = User.query.all() # All records
user = User.query.first() # First record
count = User.query.count() # Count
# Filtering
users = User.query.filter_by(is_active=True).all()
users = User.query.filter(User.is_active == True).all()
users = User.query.filter(User.name.like('%John%')).all()
# Ordering
users = User.query.order_by(User.created_at.desc()).all()
users = User.query.order_by(User.name, User.created_at).all()
# Pagination
page = 1
per_page = 20
users = User.query.paginate(page=page, per_page=per_page)
for user in users.items:
print(user.username)
# Pagination info
print(f"Page {users.page} of {users.pages}")
print(f"Total: {users.total}")
print(f"Has next: {users.has_next}")
print(f"Has prev: {users.has_prev}")
# Joins
posts = Post.query.join(User).filter(User.is_active == True).all()
# Aggregation
from sqlalchemy import func
user_count_by_date = db.session.query(
func.date(User.created_at).label('date'),
func.count(User.id).label('count')
).group_by('date').all()
# CRUD operations
# Create
user = User(email='[email protected]', username='user')
user.set_password('password')
db.session.add(user)
db.session.commit()
# Read
user = User.query.filter_by(email='[email protected]').first()
# Update
user.name = 'New Name'
db.session.commit()
# Delete
db.session.delete(user)
db.session.commit()
# Bulk insert
users = [
User(email=f'user{i}@example.com', username=f'user{i}')
for i in range(100)
]
db.session.bulk_save_objects(users)
db.session.commit()
# Transaction management
try:
user = User(email='[email protected]')
db.session.add(user)
profile = Profile(user=user, bio='Test bio')
db.session.add(profile)
db.session.commit()
except Exception as e:
db.session.rollback()
raise
# Raw SQL
from sqlalchemy import text
result = db.session.execute(
text("SELECT * FROM users WHERE email = :email"),
{"email": "[email protected]"}
)
user = result.fetchone()
Database Migrations
# Flask-Migrate setup
from flask_migrate import Migrate
migrate = Migrate(app, db)
# Initialize migrations
flask db init
# Create migration
flask db migrate -m "Initial migration"
# Apply migration
flask db upgrade
# Rollback migration
flask db downgrade
# Show migration history
flask db history
# Show current revision
flask db current
# Manual migration file
# migrations/versions/xxx_add_user_role.py
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column(
'users',
sa.Column('role', sa.String(20), nullable=True)
)
op.create_index('ix_users_role', 'users', ['role'])
def downgrade():
op.drop_index('ix_users_role', 'users')
op.drop_column('users', 'role')
# Data migration
def upgrade():
# Add column
op.add_column('users', sa.Column('email_verified', sa.Boolean))
# Set default values
op.execute("UPDATE users SET email_verified = FALSE")
# Make non-nullable
op.alter_column('users', 'email_verified', nullable=False)
# Migration with SQLAlchemy
from sqlalchemy.orm import Session
def upgrade():
bind = op.get_bind()
session = Session(bind=bind)
# Complex data transformation
users = session.execute("SELECT id, old_field FROM users")
for user in users:
new_value = transform(user.old_field)
session.execute(
"UPDATE users SET new_field = :val WHERE id = :id",
{"val": new_value, "id": user.id}
)
session.commit()
Authentication & Authorization
Flask-Login
# Setup Flask-Login
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# User model with Flask-Login
class User(UserMixin, db.Model):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True)
password_hash = db.Column(db.String(128))
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
# Login route
from flask_login import login_user, current_user
@auth_bp.route('/login', methods=['POST'])
def login():
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if user is None or not user.check_password(data['password']):
return jsonify({"error": "Invalid credentials"}), 401
login_user(user, remember=data.get('remember', False))
return jsonify({"message": "Logged in successfully"})
# Logout route
from flask_login import logout_user
@auth_bp.route('/logout', methods=['POST'])
@login_required
def logout():
logout_user()
return jsonify({"message": "Logged out successfully"})
# Protected route
from flask_login import login_required, current_user
@app.route('/profile')
@login_required
def profile():
return jsonify(current_user.to_dict())
# JWT Authentication
from flask_jwt_extended import (
JWTManager,
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity
)
jwt = JWTManager(app)
@auth_bp.route('/login', methods=['POST'])
def login():
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
return jsonify({"error": "Invalid credentials"}), 401
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
"access_token": access_token,
"refresh_token": refresh_token
})
@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
user_id = get_jwt_identity()
access_token = create_access_token(identity=user_id)
return jsonify({"access_token": access_token})
@app.route('/protected')
@jwt_required()
def protected():
user_id = get_jwt_identity()
user = User.query.get(user_id)
return jsonify(user.to_dict())
# Custom decorators
from functools import wraps
from flask import abort
def admin_required(f):
@wraps(f)
@login_required
def decorated_function(*args, **kwargs):
if not current_user.is_admin:
abort(403)
return f(*args, **kwargs)
return decorated_function
@app.route('/admin/dashboard')
@admin_required
def admin_dashboard():
return jsonify({"message": "Admin dashboard"})
# Permission decorator
def permission_required(permission):
def decorator(f):
@wraps(f)
@login_required
def decorated_function(*args, **kwargs):
if not current_user.has_permission(permission):
abort(403)
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route('/posts', methods=['POST'])
@permission_required('create_post')
def create_post():
pass
Resources & Learning Path
Learning Progression
Phase 1: Flask Fundamentals (1-2 weeks) □ Basic routing and views □ Templates with Jinja2 □ Forms and validation □ Request/response handling □ Static files Phase 2: Intermediate Flask (2-3 weeks) □ Blueprints □ Database with SQLAlchemy □ Authentication □ RESTful APIs □ Error handling Phase 3: Advanced Flask (3-4 weeks) □ Application factory pattern □ Flask extensions □ Testing strategies □ Background tasks □ WebSockets Phase 4: Production Flask (Ongoing) □ Deployment strategies □ Performance optimization □ Security hardening □ Monitoring and logging □ Microservices patterns
Related Comprehensive Sheets
Python Ecosystem → Python Advanced (decorators, context managers) → Django Production (alternative framework) → FastAPI Advanced (modern async framework) → SQLAlchemy Advanced → Celery & Task Queues Web Development → REST API Design → WebSockets → GraphQL → Authentication Patterns → API Security Testing & Quality → Pytest Advanced → Integration Testing → Load Testing → API Testing Deployment → Docker & Kubernetes → CI/CD Pipelines → Nginx Configuration → Monitoring & Observability
Pro Tips Summary
Architecture ✓ Use application factory pattern ✓ Organize code with blueprints ✓ Separate configuration by environment ✓ Keep business logic in services ✓ Use extensions for common functionality Database ✓ Use migrations for schema changes ✓ Index frequently queried columns ✓ Use eager loading to avoid N+1 ✓ Implement proper transaction handling ✓ Use connection pooling Security ✓ Never commit secrets to version control ✓ Use environment variables ✓ Implement proper authentication ✓ Validate all user input ✓ Use HTTPS in production ✓ Implement rate limiting Performance ✓ Use caching strategically ✓ Optimize database queries ✓ Enable compression ✓ Use CDN for static files ✓ Profile and monitor performance Testing ✓ Write unit tests for all routes ✓ Test error conditions ✓ Use fixtures for test data ✓ Mock external services ✓ Measure code coverage