What is Flask? The Python Micro-Framework That Gets Out of Your Way
Flask started as an April Fools joke and became one of the most widely deployed Python web frameworks in production. Here is how it works, what makes it different from Django, and what a real production Flask API looks like.
Django has an opinion about everything. Where your models go, how your URLs are structured, what your admin interface looks like, how authentication works. When you need those opinions, Django is a productivity machine. When they conflict with your requirements, Django becomes something you fight.
Flask has an opinion about almost nothing. It gives you URL routing and request handling. Everything else, you choose. Your database layer, your authentication, your template engine, your validation, your caching. Flask does not care. You compose the stack yourself.
This is not a weakness. For microservices, lightweight APIs, applications with unusual data requirements, and teams that want to make their own architectural decisions, Flask's lack of opinions is precisely the point. The framework gets out of the way so you can focus on what you are actually building.
Flask powers an enormous range of production systems. LinkedIn used it for internal services. Netflix uses Flask for some of their infrastructure tooling. Pinterest ran parts of their backend on Flask. Reddit has Flask-based services in their infrastructure. Twilio's API is built on Flask. The framework handles production traffic at scale, daily.
The Origin
Armin Ronacher created Flask in 2010. Ronacher is the same developer who created Jinja2 (the template engine), Werkzeug (the WSGI utility library), and Click (the CLI framework). Flask started as an April Fools joke: a "microframework" that fit in a single file and was too small to be taken seriously.
The joke became a real project because developers actually wanted it. They did not need a framework with built-in ORM, admin panel, and authentication for every project. They needed something that handled HTTP routing and got out of the way.
Flask 1.0 shipped in 2018, eight years after the first release. The slow version bump reflects the project's philosophy: stability over churn. Flask 2.0 arrived in 2021 with async support. Flask 3.0 shipped in 2023. The API is deliberately stable. Code written for Flask 0.12 from 2016 runs on Flask 3.x with minimal changes.
What Flask Actually Is
Flask is built on two libraries: Werkzeug and Jinja2. Both are Ronacher's work.
Werkzeug is a WSGI utility library. WSGI (Web Server Gateway Interface) is the Python standard that defines how web servers talk to Python web applications. When a request hits your server, the WSGI server (Gunicorn, uWSGI, Waitress) calls your Flask application as a function, passing request data. Your application returns a response. That is the entire protocol.
Jinja2 is a template engine. You write HTML templates with {{ variable }} syntax for substitution and {% for item in items %} syntax for logic. Flask renders templates by passing context variables to Jinja2. If you build JSON APIs instead of server-rendered HTML, you never touch Jinja2 at all.
Flask adds URL routing on top of Werkzeug. You decorate Python functions with @app.route('/path') and Flask calls the right function when a request matches that path. That is the core of what Flask provides.
A Minimal Flask Application
from flask import Flask
app = Flask(__name__)
@app.route('/health')
def health():
return {'status': 'ok'}
if __name__ == '__main__':
app.run()
Run it:
pip install flask
python app.py
Visit http://localhost:5000/health and you get {"status": "ok"}. That is the entire application. No project structure, no migration system, no configuration files. One file, one function, one route.
This is what Flask's minimal surface area looks like. Beginners get something running immediately. But this minimal example also demonstrates the trap: leaving your Flask application structured like this past the prototype stage creates maintainability problems. Production Flask applications need the application factory pattern, blueprints, and proper configuration management. The framework does not enforce these patterns, so you have to know to apply them.
The Application Factory Pattern
The application factory pattern is the correct way to structure a Flask application for production. You define a create_app() function that builds and returns a Flask app instance. This enables testing with different configurations, running multiple app instances in the same process, and avoiding circular imports.
Here is a complete production Flask project structure:
myapp/
__init__.py
config.py
models.py
extensions.py
api/
__init__.py
users.py
products.py
auth.py
utils/
__init__.py
decorators.py
validators.py
tests/
conftest.py
test_users.py
test_products.py
.env
requirements.txt
wsgi.py
Configuration:
import os
from datetime import timedelta
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-key-change-this'
SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
BCRYPT_LOG_ROUNDS = 13
class DevelopmentConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'postgresql://localhost/myapp_dev'
BCRYPT_LOG_ROUNDS = 4
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/myapp_test'
JWT_ACCESS_TOKEN_EXPIRES = timedelta(seconds=5)
BCRYPT_LOG_ROUNDS = 4
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10,
'pool_recycle': 300,
'pool_pre_ping': True,
'max_overflow': 20
}
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
Extensions module:
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from flask_bcrypt import Bcrypt
db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
bcrypt = Bcrypt()
The application factory:
from flask import Flask
from .config import config
from .extensions import db, migrate, jwt, bcrypt
def create_app(config_name: str = 'default') -> Flask:
app = Flask(__name__)
app.config.from_object(config[config_name])
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
bcrypt.init_app(app)
from .api.auth import auth_bp
from .api.users import users_bp
from .api.products import products_bp
app.register_blueprint(auth_bp, url_prefix='/api/v1/auth')
app.register_blueprint(users_bp, url_prefix='/api/v1/users')
app.register_blueprint(products_bp, url_prefix='/api/v1/products')
register_error_handlers(app)
return app
def register_error_handlers(app: Flask) -> None:
from flask import jsonify
@app.errorhandler(400)
def bad_request(error):
return jsonify({'error': 'Bad request', 'message': str(error)}), 400
@app.errorhandler(401)
def unauthorized(error):
return jsonify({'error': 'Unauthorized', 'message': str(error)}), 401
@app.errorhandler(403)
def forbidden(error):
return jsonify({'error': 'Forbidden', 'message': str(error)}), 403
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': 'Not found', 'message': str(error)}), 404
@app.errorhandler(409)
def conflict(error):
return jsonify({'error': 'Conflict', 'message': str(error)}), 409
@app.errorhandler(422)
def unprocessable(error):
return jsonify({'error': 'Unprocessable entity', 'message': str(error)}), 422
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return jsonify({'error': 'Internal server error'}), 500
wsgi.py is the entry point for production servers:
import os
from myapp import create_app
app = create_app(os.environ.get('FLASK_ENV', 'production'))
if __name__ == '__main__':
app.run()
Extensions are initialized outside the app factory in extensions.py and then bound to the app in create_app() using init_app(). This decouples the extensions from any specific app instance, which is what makes testing with multiple app configurations work correctly. If you initialize extensions directly on a module-level app = Flask(__name__), you cannot create multiple app instances without conflicts.
Models with SQLAlchemy
Flask-SQLAlchemy wraps SQLAlchemy, the most feature-complete ORM in the Python ecosystem. SQLAlchemy has more power than Django's ORM, with a steeper learning curve.
from datetime import datetime
from .extensions import db, bcrypt
class TimestampMixin:
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow,
onupdate=datetime.utcnow)
class User(TimestampMixin, db.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(255), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(255), nullable=False)
is_active = db.Column(db.Boolean, nullable=False, default=True)
is_admin = db.Column(db.Boolean, nullable=False, default=False)
last_login = db.Column(db.DateTime, nullable=True)
products = db.relationship('Product', back_populates='owner',
lazy='dynamic', cascade='all, delete-orphan')
orders = db.relationship('Order', back_populates='user',
lazy='dynamic', cascade='all, delete-orphan')
def set_password(self, password: str) -> None:
self.password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
def check_password(self, password: str) -> bool:
return bcrypt.check_password_hash(self.password_hash, password)
def to_dict(self) -> dict:
return {
'id': self.id,
'username': self.username,
'email': self.email,
'is_active': self.is_active,
'created_at': self.created_at.isoformat(),
}
def __repr__(self):
return f'<User {self.username}>'
class Category(db.Model):
__tablename__ = 'categories'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), unique=True, nullable=False)
slug = db.Column(db.String(100), unique=True, nullable=False, index=True)
description = db.Column(db.Text, nullable=True)
products = db.relationship('Product', back_populates='category', lazy='dynamic')
def to_dict(self) -> dict:
return {
'id': self.id,
'name': self.name,
'slug': self.slug,
'description': self.description,
}
class Product(TimestampMixin, db.Model):
__tablename__ = 'products'
__table_args__ = (
db.Index('ix_products_category_status', 'category_id', 'status'),
db.Index('ix_products_status_created', 'status', 'created_at'),
)
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(255), nullable=False)
slug = db.Column(db.String(255), unique=True, nullable=False, index=True)
description = db.Column(db.Text, nullable=False)
price = db.Column(db.Numeric(10, 2), nullable=False)
stock = db.Column(db.Integer, nullable=False, default=0)
status = db.Column(db.String(20), nullable=False, default='draft',
index=True)
category_id = db.Column(db.Integer, db.ForeignKey('categories.id',
ondelete='RESTRICT'), nullable=False)
owner_id = db.Column(db.Integer, db.ForeignKey('users.id',
ondelete='CASCADE'), nullable=False)
category = db.relationship('Category', back_populates='products')
owner = db.relationship('User', back_populates='products')
order_items = db.relationship('OrderItem', back_populates='product',
lazy='dynamic')
@property
def in_stock(self) -> bool:
return self.stock > 0
def reserve_stock(self, quantity: int) -> bool:
if self.stock < quantity:
return False
self.stock -= quantity
return True
def to_dict(self, include_description: bool = False) -> dict:
data = {
'id': self.id,
'name': self.name,
'slug': self.slug,
'price': float(self.price),
'stock': self.stock,
'in_stock': self.in_stock,
'status': self.status,
'category': self.category.to_dict() if self.category else None,
'created_at': self.created_at.isoformat(),
}
if include_description:
data['description'] = self.description
return data
class Order(TimestampMixin, db.Model):
__tablename__ = 'orders'
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='CASCADE'),
nullable=False, index=True)
status = db.Column(db.String(20), nullable=False, default='pending', index=True)
total_amount = db.Column(db.Numeric(10, 2), nullable=False)
shipping_address = db.Column(db.JSON, nullable=False)
user = db.relationship('User', back_populates='orders')
items = db.relationship('OrderItem', back_populates='order',
cascade='all, delete-orphan')
def to_dict(self) -> dict:
return {
'id': self.id,
'status': self.status,
'total_amount': float(self.total_amount),
'shipping_address': self.shipping_address,
'items': [item.to_dict() for item in self.items],
'created_at': self.created_at.isoformat(),
}
class OrderItem(db.Model):
__tablename__ = 'order_items'
__table_args__ = (
db.UniqueConstraint('order_id', 'product_id', name='uq_order_product'),
)
id = db.Column(db.Integer, primary_key=True)
order_id = db.Column(db.Integer, db.ForeignKey('orders.id', ondelete='CASCADE'),
nullable=False)
product_id = db.Column(db.Integer, db.ForeignKey('products.id', ondelete='RESTRICT'),
nullable=False)
quantity = db.Column(db.Integer, nullable=False)
unit_price = db.Column(db.Numeric(10, 2), nullable=False)
order = db.relationship('Order', back_populates='items')
product = db.relationship('Product', back_populates='order_items')
@property
def subtotal(self) -> float:
return float(self.quantity * self.unit_price)
def to_dict(self) -> dict:
return {
'id': self.id,
'product_id': self.product_id,
'product_name': self.product.name if self.product else None,
'quantity': self.quantity,
'unit_price': float(self.unit_price),
'subtotal': self.subtotal,
}
TimestampMixin is a reusable class that adds created_at and updated_at columns. Any model that needs timestamps inherits from it. This is composition over repetition.
lazy='dynamic' on relationships returns a query object instead of loading all related records immediately. user.orders returns a query you can filter and paginate rather than a list of every order the user has ever placed. On users with hundreds of orders, the difference between loading all records and loading a filtered page is the difference between a fast API response and a slow one.
__table_args__ with compound indexes covers query patterns that single-column indexes cannot. ix_products_category_status covers queries that filter by both category and status, which is the typical product listing query.
Blueprints
Blueprints let you split a Flask application into modular components. Each blueprint has its own routes, error handlers, and static files.
Authentication blueprint:
from flask import Blueprint, request, jsonify
from flask_jwt_extended import (
create_access_token, create_refresh_token,
jwt_required, get_jwt_identity, get_jwt
)
from datetime import datetime
from ..extensions import db
from ..models import User
auth_bp = Blueprint('auth', __name__)
REVOKED_TOKENS = set()
@auth_bp.post('/register')
def register():
data = request.get_json(silent=True)
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
required_fields = ['username', 'email', 'password']
missing = [f for f in required_fields if not data.get(f)]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
if len(data['password']) < 8:
return jsonify({'error': 'Password must be at least 8 characters'}), 400
if User.query.filter(
(User.email == data['email']) | (User.username == data['username'])
).first():
return jsonify({'error': 'Username or email already taken'}), 409
user = User(username=data['username'], email=data['email'])
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
'user': user.to_dict(),
'access_token': access_token,
'refresh_token': refresh_token,
}), 201
@auth_bp.post('/login')
def login():
data = request.get_json(silent=True)
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
user = User.query.filter(
(User.email == data.get('login')) | (User.username == data.get('login'))
).first()
if not user or not user.check_password(data.get('password', '')):
return jsonify({'error': 'Invalid credentials'}), 401
if not user.is_active:
return jsonify({'error': 'Account is disabled'}), 403
user.last_login = datetime.utcnow()
db.session.commit()
access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)
return jsonify({
'user': user.to_dict(),
'access_token': access_token,
'refresh_token': refresh_token,
})
@auth_bp.post('/refresh')
@jwt_required(refresh=True)
def refresh():
identity = get_jwt_identity()
user = User.query.get(identity)
if not user or not user.is_active:
return jsonify({'error': 'User not found or inactive'}), 401
access_token = create_access_token(identity=identity)
return jsonify({'access_token': access_token})
@auth_bp.delete('/logout')
@jwt_required()
def logout():
jti = get_jwt()['jti']
REVOKED_TOKENS.add(jti)
return jsonify({'message': 'Successfully logged out'})
@auth_bp.get('/me')
@jwt_required()
def get_current_user():
user_id = get_jwt_identity()
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
@auth_bp.post('/register') is shorthand for @auth_bp.route('/register', methods=['POST']). Flask 2.0 added get(), post(), put(), patch(), and delete() decorators. Use them.
request.get_json(silent=True) parses the request body as JSON and returns None if parsing fails instead of raising an exception. silent=True is essential in production: without it, a client sending malformed JSON triggers an unhandled exception.
Products blueprint with filtering and pagination:
from flask import Blueprint, request, jsonify, abort
from flask_jwt_extended import jwt_required, get_jwt_identity
from sqlalchemy import func, or_
from ..extensions import db
from ..models import Product, Category, User
products_bp = Blueprint('products', __name__)
def get_pagination_params():
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
return page, per_page
@products_bp.get('/')
def list_products():
page, per_page = get_pagination_params()
query = Product.query.filter_by(status='active')
category_slug = request.args.get('category')
if category_slug:
query = query.join(Category).filter(Category.slug == category_slug)
search = request.args.get('search')
if search:
search_term = f'%{search}%'
query = query.filter(
or_(
Product.name.ilike(search_term),
Product.description.ilike(search_term)
)
)
min_price = request.args.get('min_price', type=float)
max_price = request.args.get('max_price', type=float)
if min_price is not None:
query = query.filter(Product.price >= min_price)
if max_price is not None:
query = query.filter(Product.price <= max_price)
in_stock = request.args.get('in_stock', type=bool)
if in_stock:
query = query.filter(Product.stock > 0)
sort = request.args.get('sort', 'created_at')
order = request.args.get('order', 'desc')
sort_column = getattr(Product, sort, Product.created_at)
query = query.order_by(sort_column.desc() if order == 'desc' else sort_column.asc())
query = query.join(Category)
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'products': [p.to_dict() for p in pagination.items],
'pagination': {
'page': pagination.page,
'per_page': pagination.per_page,
'total': pagination.total,
'pages': pagination.pages,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev,
}
})
@products_bp.get('/<slug>')
def get_product(slug: str):
product = Product.query.filter_by(slug=slug, status='active').first_or_404()
return jsonify(product.to_dict(include_description=True))
@products_bp.post('/')
@jwt_required()
def create_product():
user_id = get_jwt_identity()
user = User.query.get_or_404(user_id)
data = request.get_json(silent=True)
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
required = ['name', 'description', 'price', 'category_id']
missing = [f for f in required if not data.get(f)]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
category = Category.query.get(data['category_id'])
if not category:
return jsonify({'error': 'Category not found'}), 404
try:
price = float(data['price'])
if price <= 0:
raise ValueError
except (TypeError, ValueError):
return jsonify({'error': 'Price must be a positive number'}), 400
from slugify import slugify
import uuid
base_slug = slugify(data['name'])
slug = base_slug
if Product.query.filter_by(slug=slug).first():
slug = f'{base_slug}-{uuid.uuid4().hex[:8]}'
product = Product(
name=data['name'],
slug=slug,
description=data['description'],
price=price,
stock=int(data.get('stock', 0)),
category_id=data['category_id'],
owner_id=user_id,
status=data.get('status', 'draft')
)
db.session.add(product)
db.session.commit()
return jsonify(product.to_dict(include_description=True)), 201
@products_bp.patch('/<slug>')
@jwt_required()
def update_product(slug: str):
user_id = get_jwt_identity()
product = Product.query.filter_by(slug=slug).first_or_404()
user = User.query.get(user_id)
if product.owner_id != user_id and not user.is_admin:
abort(403)
data = request.get_json(silent=True)
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
allowed_fields = ['name', 'description', 'price', 'stock', 'status', 'category_id']
for field in allowed_fields:
if field in data:
if field == 'price':
try:
val = float(data[field])
if val <= 0:
raise ValueError
setattr(product, field, val)
except (TypeError, ValueError):
return jsonify({'error': 'Price must be a positive number'}), 400
elif field == 'category_id':
if not Category.query.get(data[field]):
return jsonify({'error': 'Category not found'}), 404
product.category_id = data[field]
else:
setattr(product, field, data[field])
db.session.commit()
return jsonify(product.to_dict(include_description=True))
@products_bp.delete('/<slug>')
@jwt_required()
def delete_product(slug: str):
user_id = get_jwt_identity()
product = Product.query.filter_by(slug=slug).first_or_404()
user = User.query.get(user_id)
if product.owner_id != user_id and not user.is_admin:
abort(403)
db.session.delete(product)
db.session.commit()
return '', 204
first_or_404() returns the first matching record or raises a 404 error, which the error handler registered in create_app() catches and formats as JSON. This pattern eliminates the repetitive if not product: abort(404) pattern that clutters view functions.
paginate(page=page, per_page=per_page, error_out=False) handles pagination at the database level. error_out=False returns an empty page instead of raising a 404 when the page number exceeds the total pages. This is the correct behavior for paginated API endpoints: an empty result set is not an error.
setattr(product, field, data[field]) updates a model attribute by name. Combined with the allowed_fields list, this creates a flexible update endpoint that only touches explicitly permitted fields.
Testing Flask Applications
Flask's application factory pattern makes testing straightforward. You create an app instance with the testing configuration and use Flask's test client to make requests.
import pytest
import json
from myapp import create_app
from myapp.extensions import db as _db
@pytest.fixture(scope='session')
def app():
app = create_app('testing')
with app.app_context():
_db.create_all()
yield app
_db.drop_all()
@pytest.fixture(scope='function')
def db(app):
with app.app_context():
yield _db
_db.session.rollback()
for table in reversed(_db.metadata.sorted_tables):
_db.session.execute(table.delete())
_db.session.commit()
@pytest.fixture(scope='function')
def client(app):
return app.test_client()
@pytest.fixture
def auth_headers(client, db):
response = client.post('/api/v1/auth/register', json={
'username': 'testuser',
'email': '[email protected]',
'password': 'securepassword123'
})
data = response.get_json()
return {'Authorization': f'Bearer {data["access_token"]}'}
class TestAuth:
def test_register_success(self, client, db):
response = client.post('/api/v1/auth/register', json={
'username': 'newuser',
'email': '[email protected]',
'password': 'password123'
})
assert response.status_code == 201
data = response.get_json()
assert data['user']['username'] == 'newuser'
assert 'access_token' in data
assert 'refresh_token' in data
def test_register_duplicate_email(self, client, db):
payload = {
'username': 'user1',
'email': '[email protected]',
'password': 'password123'
}
client.post('/api/v1/auth/register', json=payload)
response = client.post('/api/v1/auth/register', json={
**payload,
'username': 'user2'
})
assert response.status_code == 409
def test_login_success(self, client, db):
client.post('/api/v1/auth/register', json={
'username': 'logintest',
'email': '[email protected]',
'password': 'password123'
})
response = client.post('/api/v1/auth/login', json={
'login': 'logintest',
'password': 'password123'
})
assert response.status_code == 200
assert 'access_token' in response.get_json()
def test_login_wrong_password(self, client, db):
client.post('/api/v1/auth/register', json={
'username': 'wrongpass',
'email': '[email protected]',
'password': 'correctpassword'
})
response = client.post('/api/v1/auth/login', json={
'login': 'wrongpass',
'password': 'wrongpassword'
})
assert response.status_code == 401
class TestProducts:
def test_list_products_empty(self, client, db):
response = client.get('/api/v1/products/')
assert response.status_code == 200
data = response.get_json()
assert data['products'] == []
assert data['pagination']['total'] == 0
def test_create_product_authenticated(self, client, db, auth_headers):
from myapp.models import Category
from myapp.extensions import db as _db
category = Category(name='Electronics', slug='electronics')
_db.session.add(category)
_db.session.commit()
response = client.post('/api/v1/products/', json={
'name': 'Test Product',
'description': 'A test product description',
'price': 29.99,
'stock': 100,
'category_id': category.id,
'status': 'active'
}, headers=auth_headers)
assert response.status_code == 201
data = response.get_json()
assert data['name'] == 'Test Product'
assert data['price'] == 29.99
def test_create_product_unauthenticated(self, client, db):
response = client.post('/api/v1/products/', json={
'name': 'Test Product',
'description': 'Description',
'price': 29.99,
'category_id': 1
})
assert response.status_code == 401
Each test function gets a fresh database state because the db fixture rolls back the session and deletes all rows after each test. Tests do not interfere with each other. The test client makes HTTP requests against your actual Flask routes, so you test the full request-response cycle including middleware, authentication, and error handling.
Database Migrations
flask db init
flask db migrate -m "Initial migration"
flask db upgrade
Flask-Migrate wraps Alembic, the migration tool from the SQLAlchemy team. flask db migrate inspects your models, compares them to the current schema, and generates a migration file. flask db upgrade applies pending migrations.
For production deployments:
flask db upgrade
gunicorn wsgi:app --workers 4 --bind 0.0.0.0:8000
Always run migrations before starting new application instances. The migration runs once. Application instances can scale horizontally.
Production Deployment
Flask runs on a WSGI server in production. Gunicorn is the standard choice:
pip install gunicorn gevent
gunicorn wsgi:app \
--workers 4 \
--worker-class gevent \
--worker-connections 1000 \
--bind 0.0.0.0:8000 \
--max-requests 1000 \
--max-requests-jitter 100 \
--timeout 30 \
--keep-alive 5 \
--access-logfile - \
--error-logfile -
--worker-class gevent uses greenlet-based cooperative multitasking for handling concurrent requests within each worker. Standard Flask is synchronous: one worker handles one request at a time. With gevent, each worker can handle hundreds of concurrent requests efficiently by yielding during I/O operations.
For a compute-bound application with no significant I/O, use sync workers and more of them. For an I/O-bound API service making database calls and external HTTP requests, gevent workers handle concurrency better with fewer processes.
The Nginx configuration from the Django article applies identically to Flask. Nginx handles TLS termination, serves static files, and proxies dynamic requests to Gunicorn.
Flask Application and Request Contexts
Flask has two context objects that beginners often misunderstand and senior developers sometimes mishandle. Understanding them prevents a category of subtle bugs.
The application context exists for the duration of a request or a CLI command. It provides access to current_app and g. current_app is a proxy to the Flask application that is currently handling the request. g is a request-local storage object. You put data on g at the start of a request and read it throughout. It is cleaned up at the end of every request.
The request context provides access to request (the current HTTP request object) and session (the current user's session cookie).
The common mistake is trying to use these objects outside an active context:
from flask import Flask, current_app, g, request
from functools import wraps
import time
app = Flask(__name__)
def require_api_key(f):
@wraps(f)
def decorated(*args, **kwargs):
api_key = request.headers.get('X-API-Key')
if not api_key:
return {'error': 'API key required'}, 401
from myapp.models import APIKey
key_record = APIKey.query.filter_by(key=api_key, is_active=True).first()
if not key_record:
return {'error': 'Invalid API key'}, 401
g.api_client = key_record.client
g.api_key_id = key_record.id
return f(*args, **kwargs)
return decorated
def log_request_timing(f):
@wraps(f)
def decorated(*args, **kwargs):
g.request_start = time.monotonic()
response = f(*args, **kwargs)
elapsed = time.monotonic() - g.request_start
current_app.logger.info(
f"{request.method} {request.path} completed in {elapsed:.3f}s"
)
return response
return decorated
@app.before_request
def before_request():
g.request_start = time.monotonic()
@app.after_request
def after_request(response):
if hasattr(g, 'request_start'):
elapsed = time.monotonic() - g.request_start
response.headers['X-Response-Time'] = f"{elapsed:.3f}s"
return response
@app.teardown_appcontext
def teardown_db(exception=None):
db = g.pop('db', None)
if db is not None:
db.close()
@app.before_request runs before every request. @app.after_request runs after every request and receives the response object, which you can modify before it is sent. @app.teardown_appcontext runs during teardown even if an exception occurred. Use it to clean up resources that must be released regardless of request outcome.
The context objects only exist inside a request or explicit with app.app_context(): block. Running background threads or Celery tasks requires pushing an application context manually:
def run_background_task(task_func, *args):
with app.app_context():
task_func(*args)
Celery handles this automatically when you use @shared_task within a Flask application that has Celery configured. But custom threads need the context push explicitly.
Background Tasks with Celery
import os
from celery import Celery
def make_celery(app):
celery = Celery(
app.import_name,
broker=app.config['CELERY_BROKER_URL'],
backend=app.config['CELERY_RESULT_BACKEND']
)
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
In create_app():
from .celery_utils import make_celery
celery = None
def create_app(config_name='default'):
global celery
app = Flask(__name__)
app.config.from_object(config[config_name])
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
bcrypt.init_app(app)
celery = make_celery(app)
from .api.auth import auth_bp
from .api.users import users_bp
from .api.products import products_bp
app.register_blueprint(auth_bp, url_prefix='/api/v1/auth')
app.register_blueprint(users_bp, url_prefix='/api/v1/users')
app.register_blueprint(products_bp, url_prefix='/api/v1/products')
register_error_handlers(app)
return app
Tasks:
from . import celery
from .extensions import db
from .models import User, Order
import requests
import logging
logger = logging.getLogger(__name__)
@celery.task(
bind=True,
max_retries=3,
default_retry_delay=60,
autoretry_for=(Exception,),
retry_backoff=True
)
def send_order_confirmation(self, order_id: int) -> dict:
order = Order.query.options(
db.joinedload(Order.user),
db.joinedload(Order.items)
).get(order_id)
if not order:
logger.warning(f"Order {order_id} not found for confirmation email")
return {'status': 'skipped'}
from flask_mail import Message
from flask import current_app, render_template
msg = Message(
subject=f'Order Confirmation #{order.id}',
recipients=[order.user.email],
html=render_template('emails/order_confirmation.html', order=order)
)
from . import mail
mail.send(msg)
logger.info(f"Order confirmation sent for order {order_id}")
return {'status': 'sent', 'order_id': order_id}
@celery.task(
bind=True,
max_retries=5,
time_limit=300,
soft_time_limit=240
)
def sync_inventory_from_supplier(self, supplier_id: int) -> dict:
from .models import Supplier, Product
supplier = Supplier.query.get(supplier_id)
if not supplier:
return {'status': 'skipped', 'reason': 'supplier_not_found'}
try:
response = requests.get(
f"{supplier.api_url}/inventory",
headers={'Authorization': f'Bearer {supplier.api_key}'},
timeout=30
)
response.raise_for_status()
inventory_data = response.json()
except requests.RequestException as exc:
logger.error(f"Failed to fetch inventory from supplier {supplier_id}: {exc}")
raise self.retry(exc=exc, countdown=120)
updated_count = 0
for item in inventory_data['items']:
product = Product.query.filter_by(
supplier_sku=item['sku'],
owner_id=supplier.owner_id
).first()
if product:
product.stock = item['quantity']
updated_count += 1
db.session.commit()
logger.info(f"Updated {updated_count} products from supplier {supplier_id}")
return {'status': 'success', 'updated': updated_count}
The ContextTask base class pushes an application context before every task execution. This makes db, current_app, render_template, and every other Flask context-dependent feature work inside Celery tasks without manual context management.
Async Flask
Flask 2.0 added native async support. You can write async view functions and use await inside them:
import asyncio
import aiohttp
from flask import Flask, jsonify
app = Flask(__name__)
@app.get('/data/aggregate')
async def aggregate_data():
async with aiohttp.ClientSession() as session:
tasks = [
fetch_service_data(session, 'https://service-a.internal/data'),
fetch_service_data(session, 'https://service-b.internal/data'),
fetch_service_data(session, 'https://service-c.internal/data'),
]
results = await asyncio.gather(*tasks, return_exceptions=True)
aggregated = {
'service_a': results[0] if not isinstance(results[0], Exception) else None,
'service_b': results[1] if not isinstance(results[1], Exception) else None,
'service_c': results[2] if not isinstance(results[2], Exception) else None,
}
return jsonify(aggregated)
async def fetch_service_data(session: aiohttp.ClientSession, url: str) -> dict:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
response.raise_for_status()
return await response.json()
The caveat: Flask's async support runs on Werkzeug's synchronous WSGI server unless you use an ASGI server like Hypercorn or Uvicorn with the ASGI adapter. Standard Gunicorn with sync workers does not benefit from async view functions in Flask. If async performance is your primary requirement, FastAPI running on Uvicorn with native async is the better choice. Flask's async support covers the use case where you have an existing Flask application and want to use await in specific endpoints without rewriting everything.
For ASGI deployment with Flask async:
pip install hypercorn
hypercorn wsgi:app --bind 0.0.0.0:8000 --workers 4
Structured Logging
Production applications need structured logs that monitoring systems can parse. Raw text logs do not scale past a few hundred lines of daily output.
import logging
import json
import time
from datetime import datetime
from flask import Flask, request, g
class JSONFormatter(logging.Formatter):
def format(self, record):
log_data = {
'timestamp': datetime.utcnow().isoformat(),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
}
if hasattr(record, 'request_id'):
log_data['request_id'] = record.request_id
if hasattr(record, 'user_id'):
log_data['user_id'] = record.user_id
if hasattr(record, 'duration_ms'):
log_data['duration_ms'] = record.duration_ms
if record.exc_info:
log_data['exception'] = self.formatException(record.exc_info)
return json.dumps(log_data)
def setup_logging(app: Flask) -> None:
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
app.logger.handlers.clear()
app.logger.addHandler(handler)
app.logger.setLevel(logging.INFO)
for logger_name in ('sqlalchemy.engine', 'werkzeug'):
logging.getLogger(logger_name).setLevel(logging.WARNING)
def configure_request_logging(app: Flask) -> None:
import uuid
@app.before_request
def start_timer():
g.start_time = time.monotonic()
g.request_id = str(uuid.uuid4())[:8]
@app.after_request
def log_request(response):
duration_ms = int((time.monotonic() - g.start_time) * 1000)
app.logger.info(
f"{request.method} {request.path} {response.status_code}",
extra={
'request_id': g.request_id,
'method': request.method,
'path': request.path,
'status_code': response.status_code,
'duration_ms': duration_ms,
'ip': request.remote_addr,
}
)
response.headers['X-Request-ID'] = g.request_id
response.headers['X-Response-Time'] = f"{duration_ms}ms"
return response
JSON logs flow directly into log aggregation systems like ELK (Elasticsearch, Logstash, Kibana), Datadog, or CloudWatch Logs. Each log entry carries structured metadata, so you can query "all requests from user_id 1234 with status_code 500 in the last hour" rather than grepping text files.
Flask Caching
from flask_caching import Cache
cache = Cache()
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
app.config['CACHE_TYPE'] = 'RedisCache'
app.config['CACHE_REDIS_URL'] = os.environ.get('REDIS_URL', 'redis://localhost:6379/2')
app.config['CACHE_DEFAULT_TIMEOUT'] = 300
cache.init_app(app)
return app
Usage in views:
from .extensions import cache
@products_bp.get('/featured')
@cache.cached(timeout=600, key_prefix='featured_products')
def get_featured_products():
products = Product.query.filter_by(
status='active',
is_featured=True
).order_by(Product.created_at.desc()).limit(12).all()
return jsonify([p.to_dict() for p in products])
def get_product_by_slug_cached(slug: str) -> dict | None:
cache_key = f'product_{slug}'
data = cache.get(cache_key)
if data is None:
product = Product.query.filter_by(slug=slug, status='active').first()
if not product:
return None
data = product.to_dict(include_description=True)
cache.set(cache_key, data, timeout=300)
return data
def invalidate_product_cache(slug: str) -> None:
cache.delete(f'product_{slug}')
cache.delete('featured_products')
The @cache.cached() decorator caches the entire response for a view function. key_prefix sets the cache key. If two different users call the same endpoint, they get the same cached response. For user-specific data, use cache.memoize() instead and include the user ID in the cache key.
Cache invalidation on data change:
@products_bp.patch('/<slug>')
@jwt_required()
def update_product(slug: str):
product = Product.query.filter_by(slug=slug).first_or_404()
data = request.get_json(silent=True)
for field in ['name', 'description', 'price', 'stock', 'status']:
if field in data:
setattr(product, field, data[field])
db.session.commit()
invalidate_product_cache(slug)
return jsonify(product.to_dict(include_description=True))
Rate Limiting
APIs without rate limiting get hammered. Either by careless clients in an infinite loop, or by people intentionally trying to exhaust your resources. Flask-Limiter wraps your endpoints with configurable request limits.
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
key_func=get_remote_address,
default_limits=['200 per hour', '50 per minute'],
storage_uri=os.environ.get('REDIS_URL', 'redis://localhost:6379/3')
)
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
limiter.init_app(app)
return app
Apply limits per endpoint:
@auth_bp.post('/login')
@limiter.limit('10 per minute')
@limiter.limit('50 per hour')
def login():
data = request.get_json(silent=True)
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
pass
@auth_bp.post('/register')
@limiter.limit('5 per minute')
@limiter.limit('20 per hour')
def register():
pass
@products_bp.get('/')
@limiter.limit('300 per hour')
def list_products():
pass
Login endpoints get stricter limits because credential stuffing attacks iterate through username/password combinations as fast as the server allows. A limit of 10 per minute per IP stops brute force attempts from being practical. The Redis backend stores rate limit state, so limits work correctly across multiple Gunicorn workers.
For authenticated endpoints, limit by user ID instead of IP address. A user behind a NAT might share an IP with thousands of others, and a shared IP limit would break everyone on the network when one person hits the limit:
def get_user_id_or_ip():
from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity
try:
verify_jwt_in_request(optional=True)
user_id = get_jwt_identity()
if user_id:
return f'user:{user_id}'
except Exception:
pass
return get_remote_address()
authenticated_limiter = Limiter(
key_func=get_user_id_or_ip,
storage_uri=os.environ.get('REDIS_URL', 'redis://localhost:6379/3')
)
Health Checks and Readiness Probes
Container orchestration systems (Kubernetes, ECS) need endpoints to verify that your application is alive and ready to receive traffic. A health check that only returns 200 OK is useless. A health check that verifies database connectivity and dependency availability tells the orchestrator whether to send traffic to this instance.
from flask import Blueprint, jsonify, current_app
from datetime import datetime
import time
health_bp = Blueprint('health', __name__)
@health_bp.get('/health/live')
def liveness():
return jsonify({'status': 'alive', 'timestamp': datetime.utcnow().isoformat()})
@health_bp.get('/health/ready')
def readiness():
checks = {}
overall_status = 'ready'
start = time.monotonic()
try:
from .extensions import db
db.session.execute(db.text('SELECT 1'))
checks['database'] = {
'status': 'ok',
'latency_ms': round((time.monotonic() - start) * 1000, 2)
}
except Exception as e:
checks['database'] = {'status': 'error', 'error': str(e)}
overall_status = 'not_ready'
start = time.monotonic()
try:
from .extensions import cache
cache.set('health_check', 'ok', timeout=10)
assert cache.get('health_check') == 'ok'
checks['cache'] = {
'status': 'ok',
'latency_ms': round((time.monotonic() - start) * 1000, 2)
}
except Exception as e:
checks['cache'] = {'status': 'error', 'error': str(e)}
overall_status = 'not_ready'
status_code = 200 if overall_status == 'ready' else 503
return jsonify({
'status': overall_status,
'checks': checks,
'timestamp': datetime.utcnow().isoformat(),
'version': current_app.config.get('APP_VERSION', 'unknown')
}), status_code
The liveness probe returns 200 as long as the Python process is running. Kubernetes restarts the container if the liveness probe fails. The readiness probe verifies that the application can actually serve requests: database is reachable, cache is reachable. Kubernetes stops routing traffic to the instance if the readiness probe fails but does not restart it. This distinction matters during deployments: a new instance that is still running migrations should fail readiness until migrations complete, without triggering a restart loop.
Flask vs Django vs FastAPI: The Actual Decision
The question is not which framework is better. The question is what your project needs.
Use Flask when:
- You build microservices or lightweight APIs that do not need an ORM, admin panel, or migrations out of the box.
- Your data layer is unusual. Maybe you query a graph database, or read from Redis, or call gRPC services. Django's ORM would just be in the way.
- You want explicit control over every library in your stack. Flask forces no authentication library, no ORM, no template engine.
- You need to embed a web interface in a larger Python application, like a monitoring dashboard or a data processing tool.
- Your team already knows SQLAlchemy and prefers it over Django's ORM.
Use Django when:
- Non-technical team members need to manage data through an admin interface.
- Your project has complex relational data with many models.
- You want authentication, migrations, and an ORM without configuring them.
- You are building a CMS, e-commerce platform, or content-heavy site.
- Development speed matters more than customization.
Use FastAPI when:
- You need auto-generated API documentation.
- Your codebase uses type hints throughout and you want request validation from them.
- Performance under concurrent load is a requirement.
- You build a new service from scratch and async is a first-class concern.
Flask's strength is exactly what critics call its weakness: it has no opinions. You build your stack deliberately. You know every library you add and why it is there. For experienced developers who know what they want, that freedom produces cleaner, more focused applications than any full-stack framework imposes.
Armin Ronacher built Flask as a joke. Then he built it properly because developers needed it. Twenty million downloads per month later, that April Fools project is running in production at some of the largest technology companies in the world.