Flask Expert

Flask Expert

Expert guidance for Flask web development, building REST APIs, using extensions, and production deployment.

Core Concepts

Flask Fundamentals

• Routing and views
• Request/response handling
• Templates with Jinja2
• Blueprints for modularity
• Application factory pattern
• Configuration management

Flask Extensions

• Flask-SQLAlchemy (ORM)
• Flask-Migrate (database migrations)
• Flask-Login (authentication)
• Flask-RESTful (REST APIs)
• Flask-JWT-Extended (JWT tokens)
• Flask-CORS (CORS handling)
• Flask-Limiter (rate limiting)

Best Practices

• Application structure
• Error handling
• Testing
• Security
• Production deployment

Basic Flask Application

from flask import Flask, request, jsonify, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime

# Application factory
def create_app(config_name='development'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    # Initialize extensions
    db.init_app(app)
    migrate.init_app(app, db)

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

    from .auth import auth_bp
    app.register_blueprint(auth_bp, url_prefix='/auth')

    return app

# Database setup
db = SQLAlchemy()
migrate = Migrate()

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

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    posts = db.relationship('Post', backref='author', lazy='dynamic')

    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,
            'created_at': self.created_at.isoformat()
        }

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, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'content': self.content,
            'author': self.author.to_dict(),
            'created_at': self.created_at.isoformat()
        }

REST API with Flask-RESTful

from flask import Blueprint
from flask_restful import Api, Resource, reqparse, fields, marshal_with
from flask_jwt_extended import jwt_required, get_jwt_identity
from .models import db, User, Post

api_bp = Blueprint('api', __name__)
api = Api(api_bp)

# Request parsers
user_parser = reqparse.RequestParser()
user_parser.add_argument('email', type=str, required=True, help='Email is required')
user_parser.add_argument('password', type=str, required=True, help='Password is required')

post_parser = reqparse.RequestParser()
post_parser.add_argument('title', type=str, required=True)
post_parser.add_argument('content', type=str, required=True)

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

post_fields = {
    'id': fields.Integer,
    'title': fields.String,
    'content': fields.String,
    'author': fields.Nested(user_fields),
    'created_at': fields.DateTime(dt_format='iso8601')
}

# Resources
class UserListResource(Resource):
    @marshal_with(user_fields)
    def get(self):
        """Get all users"""
        page = request.args.get('page', 1, type=int)
        per_page = request.args.get('per_page', 20, type=int)

        users = User.query.paginate(page=page, per_page=per_page)
        return users.items

    def post(self):
        """Create new user"""
        args = user_parser.parse_args()

        if User.query.filter_by(email=args['email']).first():
            return {'message': 'Email already exists'}, 400

        user = User(email=args['email'])
        user.set_password(args['password'])

        db.session.add(user)
        db.session.commit()

        return user.to_dict(), 201

class UserResource(Resource):
    @marshal_with(user_fields)
    def get(self, user_id):
        """Get user by ID"""
        user = User.query.get_or_404(user_id)
        return user

    @jwt_required()
    def delete(self, user_id):
        """Delete user"""
        current_user_id = get_jwt_identity()

        if current_user_id != user_id:
            return {'message': 'Unauthorized'}, 403

        user = User.query.get_or_404(user_id)
        db.session.delete(user)
        db.session.commit()

        return '', 204

class PostListResource(Resource):
    @marshal_with(post_fields)
    def get(self):
        """Get all posts"""
        posts = Post.query.order_by(Post.created_at.desc()).all()
        return posts

    @jwt_required()
    def post(self):
        """Create new post"""
        args = post_parser.parse_args()
        current_user_id = get_jwt_identity()

        post = Post(
            title=args['title'],
            content=args['content'],
            user_id=current_user_id
        )

        db.session.add(post)
        db.session.commit()

        return post.to_dict(), 201

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

Authentication with JWT

from flask import Blueprint, request, jsonify
from flask_jwt_extended import (
    JWTManager, create_access_token, create_refresh_token,
    jwt_required, get_jwt_identity
)
from .models import db, User

auth_bp = Blueprint('auth', __name__)
jwt = JWTManager()

@auth_bp.route('/register', methods=['POST'])
def register():
    """Register new user"""
    data = request.get_json()

    if not data or not data.get('email') or not data.get('password'):
        return jsonify({'message': 'Missing required fields'}), 400

    if User.query.filter_by(email=data['email']).first():
        return jsonify({'message': 'Email already exists'}), 400

    user = User(email=data['email'])
    user.set_password(data['password'])

    db.session.add(user)
    db.session.commit()

    return jsonify(user.to_dict()), 201

@auth_bp.route('/login', methods=['POST'])
def login():
    """Login user"""
    data = request.get_json()

    if not data or not data.get('email') or not data.get('password'):
        return jsonify({'message': 'Missing credentials'}), 400

    user = User.query.filter_by(email=data['email']).first()

    if not user or not user.check_password(data['password']):
        return jsonify({'message': '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,
        'user': user.to_dict()
    }), 200

@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    """Refresh access token"""
    current_user_id = get_jwt_identity()
    access_token = create_access_token(identity=current_user_id)

    return jsonify({'access_token': access_token}), 200

@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
    """Get current authenticated user"""
    current_user_id = get_jwt_identity()
    user = User.query.get_or_404(current_user_id)

    return jsonify(user.to_dict()), 200

Error Handling

from flask import jsonify
from sqlalchemy.exc import IntegrityError
from werkzeug.exceptions import HTTPException

def register_error_handlers(app):
    """Register error handlers"""

    @app.errorhandler(HTTPException)
    def handle_http_exception(e):
        """Handle HTTP exceptions"""
        return jsonify({
            'error': e.name,
            'message': e.description,
            'code': e.code
        }), e.code

    @app.errorhandler(IntegrityError)
    def handle_integrity_error(e):
        """Handle database integrity errors"""
        db.session.rollback()
        return jsonify({
            'error': 'Database error',
            'message': 'A database constraint was violated'
        }), 400

    @app.errorhandler(Exception)
    def handle_exception(e):
        """Handle unexpected exceptions"""
        app.logger.exception(e)
        return jsonify({
            'error': 'Internal server error',
            'message': 'An unexpected error occurred'
        }), 500

    @app.errorhandler(404)
    def not_found(e):
        """Handle 404 errors"""
        return jsonify({
            'error': 'Not found',
            'message': 'The requested resource was not found'
        }), 404

    @app.errorhandler(400)
    def bad_request(e):
        """Handle 400 errors"""
        return jsonify({
            'error': 'Bad request',
            'message': 'The request was invalid'
        }), 400

Rate Limiting

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

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

def create_app():
    app = Flask(__name__)
    limiter.init_app(app)

    @app.route('/api/search')
    @limiter.limit("10 per minute")
    def search():
        """Rate-limited search endpoint"""
        query = request.args.get('q')
        # Perform search
        return jsonify({'results': []})

    @app.route('/api/expensive')
    @limiter.limit("1 per minute")
    def expensive_operation():
        """Very rate-limited expensive operation"""
        # Expensive computation
        return jsonify({'status': 'completed'})

    return app

Configuration

import os

class Config:
    """Base configuration"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret-key'
    JWT_ACCESS_TOKEN_EXPIRES = 3600  # 1 hour
    JWT_REFRESH_TOKEN_EXPIRES = 2592000  # 30 days

class DevelopmentConfig(Config):
    """Development configuration"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'

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

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

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

Testing

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):
    """Create test client"""
    return app.test_client()

@pytest.fixture
def auth_headers(client):
    """Create authenticated headers"""
    # Register user
    client.post('/auth/register', json={
        'email': 'test@example.com',
        'password': 'password123'
    })

    # Login
    response = client.post('/auth/login', json={
        'email': 'test@example.com',
        'password': 'password123'
    })

    token = response.json['access_token']
    return {'Authorization': f'Bearer {token}'}

def test_register_user(client):
    """Test user registration"""
    response = client.post('/auth/register', json={
        'email': 'new@example.com',
        'password': 'password123'
    })

    assert response.status_code == 201
    assert response.json['email'] == 'new@example.com'

def test_login(client):
    """Test user login"""
    # Register first
    client.post('/auth/register', json={
        'email': 'test@example.com',
        'password': 'password123'
    })

    # Login
    response = client.post('/auth/login', json={
        'email': 'test@example.com',
        'password': 'password123'
    })

    assert response.status_code == 200
    assert 'access_token' in response.json

def test_create_post(client, auth_headers):
    """Test creating a post"""
    response = client.post('/api/posts',
        json={
            'title': 'Test Post',
            'content': 'Test content'
        },
        headers=auth_headers
    )

    assert response.status_code == 201
    assert response.json['title'] == 'Test Post'

def test_get_posts(client):
    """Test getting all posts"""
    response = client.get('/api/posts')

    assert response.status_code == 200
    assert isinstance(response.json, list)

Production Deployment

# wsgi.py
from app import create_app
import os

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

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

# Using Gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 wsgi:app

# With proper workers
gunicorn -w 4 \
  --worker-class gevent \
  --worker-connections 1000 \
  --timeout 30 \
  --keep-alive 5 \
  --log-level info \
  --access-logfile - \
  --error-logfile - \
  wsgi:app

Best Practices

• Use application factory pattern
• Organize code with blueprints
• Implement proper error handling
• Use environment variables for config
• Write comprehensive tests
• Use database migrations
• Implement authentication/authorization
• Add rate limiting
• Enable CORS properly
• Use connection pooling
• Log appropriately
• Monitor performance

Anti-Patterns

❌ Global app instance ❌ No database migrations ❌ Hardcoded configuration ❌ Missing error handling ❌ No authentication ❌ Exposing sensitive data ❌ Not using blueprints

Resources

• Flask Documentation: https://flask.palletsprojects.com/
• Flask-SQLAlchemy: https://flask-sqlalchemy.palletsprojects.com/
• Flask-RESTful: https://flask-restful.readthedocs.io/
• Flask-JWT-Extended: https://flask-jwt-extended.readthedocs.io/
• Miguel Grinberg's Flask Mega-Tutorial: https://blog.miguelgrinberg.com/

Converted and distributed by TomeVault — claim your Tome and manage your conversions.