SuperAPI AI Directory
Partner with us中文AI newsFavorites
Skillsflask-expert

flask-expert

Coding & development

View repository

Updated 10-16 08:03

About this skill

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()
        }

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

Reference Documentation

Detailed material lives alongside this skill and is read on demand:

Resources