flask-api-patterns

---
name: "flask-api-patterns"
description: "Master Flask RESTful API development with blueprints, extensions, and patterns for building lightweight, flexible Python web services."
license: "MIT"
---

# Flask API Patterns

Master Flask's microframework approach to building RESTful APIs with blueprints, extensions, and proven patterns for creating lightweight, flexible, and maintainable Python web services.

## When to Use This Skill

- Building RESTful APIs with Flask
- Creating microservices with Python
- Implementing API authentication and authorization
- Organizing Flask applications with blueprints
- Integrating Flask extensions
- Testing Flask applications
- Deploying Flask APIs to production

## Core Concepts

### 1. Application Factory Pattern

```python
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from flask_cors import CORS

db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()

def create_app(config_name='development'):
    app = Flask(__name__)

    # Load configuration
    app.config.from_object(config[config_name])

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

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

    # Register error handlers
    register_error_handlers(app)

    # Register CLI commands
    register_cli_commands(app)

    return app

def register_error_handlers(app):
    @app.errorhandler(400)
    def bad_request(error):
        return {'error': 'Bad request', 'message': str(error)}, 400

    @app.errorhandler(404)
    def not_found(error):
        return {'error': 'Not found', 'message': str(error)}, 404

    @app.errorhandler(500)
    def internal_error(error):
        db.session.rollback()
        return {'error': 'Internal server error'}, 500

def register_cli_commands(app):
    @app.cli.command()
    def seed():
        """Seed the database."""
        from app.seeds import run_seeds
        run_seeds()
```

### 2. Configuration Management

```python
# config.py
import os
from datetime import timedelta

class Config:
    """Base configuration."""
    SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret-key')
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)

class DevelopmentConfig(Config):
    """Development configuration."""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        'DATABASE_URL',
        'postgresql://localhost/flask_dev'
    )

class TestingConfig(Config):
    """Testing configuration."""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(seconds=5)

class ProductionConfig(Config):
    """Production configuration."""
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ['DATABASE_URL']
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(minutes=15)

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

### 3. Models with SQLAlchemy

```python
# app/models/user.py
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from app import db

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(256), nullable=False)
    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
    articles = db.relationship('Article', 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,
            'username': self.username,
            'is_active': self.is_active,
            'created_at': self.created_at.isoformat(),
        }

    def __repr__(self):
        return f'<User {self.username}>'

# app/models/article.py
class Article(db.Model):
    __tablename__ = 'articles'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    slug = db.Column(db.String(200), unique=True, nullable=False, index=True)
    body = db.Column(db.Text, nullable=False)
    status = db.Column(db.String(20), default='draft')
    author_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    published_at = db.Column(db.DateTime)

    # Relationships
    tags = db.relationship('Tag', secondary='article_tags', backref='articles')

    @staticmethod
    def generate_slug(title):
        from slugify import slugify
        return slugify(title)

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'slug': self.slug,
            'body': self.body,
            'status': self.status,
            'author': self.author.to_dict(),
            'tags': [tag.name for tag in self.tags],
            'created_at': self.created_at.isoformat(),
        }
```

## Essential Patterns

### Pattern 1: Blueprints and API Structure

```python
# app/api/v1/__init__.py
from flask import Blueprint

api_v1 = Blueprint('api_v1', __name__)

from app.api.v1 import auth, users, articles  # noqa

# app/api/v1/articles.py
from flask import request, jsonify
from flask_jwt_extended import jwt_required, current_user
from app.api.v1 import api_v1
from app.models import Article, Tag
from app.schemas import ArticleSchema, ArticleCreateSchema
from app import db

article_schema = ArticleSchema()
articles_schema = ArticleSchema(many=True)

@api_v1.route('/articles', methods=['GET'])
def get_articles():
    """Get paginated list of articles."""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    status = request.args.get('status', 'published')

    query = Article.query.filter_by(status=status)

    # Search
    search = request.args.get('q')
    if search:
        query = query.filter(
            db.or_(
                Article.title.ilike(f'%{search}%'),
                Article.body.ilike(f'%{search}%')
            )
        )

    # Tag filter
    tag = request.args.get('tag')
    if tag:
        query = query.join(Article.tags).filter(Tag.name == tag)

    pagination = query.order_by(Article.created_at.desc()).paginate(
        page=page, per_page=per_page, error_out=False
    )

    return jsonify({
        'articles': articles_schema.dump(pagination.items),
        'meta': {
            'page': page,
            'per_page': per_page,
            'total': pagination.total,
            'pages': pagination.pages,
        }
    })

@api_v1.route('/articles/<slug>', methods=['GET'])
def get_article(slug):
    """Get article by slug."""
    article = Article.query.filter_by(slug=slug).first_or_404()
    return jsonify(article_schema.dump(article))

@api_v1.route('/articles', methods=['POST'])
@jwt_required()
def create_article():
    """Create new article."""
    schema = ArticleCreateSchema()
    data = schema.load(request.json)

    article = Article(
        title=data['title'],
        slug=Article.generate_slug(data['title']),
        body=data['body'],
        status=data.get('status', 'draft'),
        author=current_user
    )

    # Handle tags
    if 'tags' in data:
        for tag_name in data['tags']:
            tag = Tag.query.filter_by(name=tag_name).first()
            if not tag:
                tag = Tag(name=tag_name)
            article.tags.append(tag)

    db.session.add(article)
    db.session.commit()

    return jsonify(article_schema.dump(article)), 201

@api_v1.route('/articles/<slug>', methods=['PUT'])
@jwt_required()
def update_article(slug):
    """Update article."""
    article = Article.query.filter_by(slug=slug).first_or_404()

    # Check ownership
    if article.author_id != current_user.id:
        return jsonify({'error': 'Forbidden'}), 403

    schema = ArticleCreateSchema(partial=True)
    data = schema.load(request.json)

    for key, value in data.items():
        if key == 'tags':
            article.tags = []
            for tag_name in value:
                tag = Tag.query.filter_by(name=tag_name).first()
                if not tag:
                    tag = Tag(name=tag_name)
                article.tags.append(tag)
        else:
            setattr(article, key, value)

    db.session.commit()
    return jsonify(article_schema.dump(article))

@api_v1.route('/articles/<slug>', methods=['DELETE'])
@jwt_required()
def delete_article(slug):
    """Delete article."""
    article = Article.query.filter_by(slug=slug).first_or_404()

    if article.author_id != current_user.id:
        return jsonify({'error': 'Forbidden'}), 403

    db.session.delete(article)
    db.session.commit()

    return '', 204
```

### Pattern 2: Marshmallow Schemas

```python
# app/schemas.py
from marshmallow import Schema, fields, validate, validates, ValidationError, post_load

class UserSchema(Schema):
    id = fields.Int(dump_only=True)
    email = fields.Email(required=True)
    username = fields.Str(required=True, validate=validate.Length(min=3, max=80))
    password = fields.Str(load_only=True, required=True, validate=validate.Length(min=8))
    is_active = fields.Bool(dump_only=True)
    created_at = fields.DateTime(dump_only=True)

    @validates('username')
    def validate_username(self, value):
        from app.models import User
        if User.query.filter_by(username=value).first():
            raise ValidationError('Username already exists.')

class UserLoginSchema(Schema):
    email = fields.Email(required=True)
    password = fields.Str(required=True)

class ArticleSchema(Schema):
    id = fields.Int(dump_only=True)
    title = fields.Str(required=True, validate=validate.Length(min=5, max=200))
    slug = fields.Str(dump_only=True)
    body = fields.Str(required=True)
    status = fields.Str(validate=validate.OneOf(['draft', 'published', 'archived']))
    author = fields.Nested(UserSchema, dump_only=True)
    tags = fields.List(fields.Str())
    created_at = fields.DateTime(dump_only=True)
    updated_at = fields.DateTime(dump_only=True)

class ArticleCreateSchema(Schema):
    title = fields.Str(required=True, validate=validate.Length(min=5, max=200))
    body = fields.Str(required=True)
    status = fields.Str(
        validate=validate.OneOf(['draft', 'published']),
        load_default='draft'
    )
    tags = fields.List(fields.Str())

class PaginationSchema(Schema):
    page = fields.Int(load_default=1, validate=validate.Range(min=1))
    per_page = fields.Int(load_default=20, validate=validate.Range(min=1, max=100))
```

### Pattern 3: Authentication with JWT

```python
# app/api/v1/auth.py
from flask import request, jsonify
from flask_jwt_extended import (
    create_access_token, create_refresh_token,
    jwt_required, get_jwt_identity, current_user
)
from app.api.v1 import api_v1
from app.models import User
from app.schemas import UserSchema, UserLoginSchema
from app import db, jwt

user_schema = UserSchema()

@jwt.user_identity_loader
def user_identity_lookup(user):
    return user.id

@jwt.user_lookup_loader
def user_lookup_callback(_jwt_header, jwt_data):
    identity = jwt_data['sub']
    return User.query.get(identity)

@api_v1.route('/auth/register', methods=['POST'])
def register():
    """Register new user."""
    schema = UserSchema()
    data = schema.load(request.json)

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

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

    access_token = create_access_token(identity=user)
    refresh_token = create_refresh_token(identity=user)

    return jsonify({
        'user': user_schema.dump(user),
        'access_token': access_token,
        'refresh_token': refresh_token
    }), 201

@api_v1.route('/auth/login', methods=['POST'])
def login():
    """Login user."""
    schema = UserLoginSchema()
    data = schema.load(request.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

    if not user.is_active:
        return jsonify({'error': 'Account is disabled'}), 401

    access_token = create_access_token(identity=user)
    refresh_token = create_refresh_token(identity=user)

    return jsonify({
        'user': user_schema.dump(user),
        'access_token': access_token,
        'refresh_token': refresh_token
    })

@api_v1.route('/auth/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    """Refresh access token."""
    access_token = create_access_token(identity=current_user)
    return jsonify({'access_token': access_token})

@api_v1.route('/auth/me', methods=['GET'])
@jwt_required()
def get_current_user():
    """Get current user."""
    return jsonify(user_schema.dump(current_user))
```

### Pattern 4: Service Layer

```python
# app/services/article_service.py
from datetime import datetime
from app import db
from app.models import Article, Tag

class ArticleService:
    @staticmethod
    def create(author, title, body, status='draft', tags=None):
        """Create a new article."""
        article = Article(
            title=title,
            slug=Article.generate_slug(title),
            body=body,
            status=status,
            author=author
        )

        if tags:
            ArticleService._attach_tags(article, tags)

        db.session.add(article)
        db.session.commit()

        return article

    @staticmethod
    def update(article, **kwargs):
        """Update an article."""
        tags = kwargs.pop('tags', None)

        for key, value in kwargs.items():
            if hasattr(article, key):
                setattr(article, key, value)

        if tags is not None:
            ArticleService._attach_tags(article, tags)

        db.session.commit()
        return article

    @staticmethod
    def publish(article):
        """Publish an article."""
        article.status = 'published'
        article.published_at = datetime.utcnow()
        db.session.commit()

        # Trigger notifications
        from app.tasks import notify_followers
        notify_followers.delay(article.id)

        return article

    @staticmethod
    def _attach_tags(article, tag_names):
        """Attach tags to article."""
        article.tags = []
        for name in tag_names:
            tag = Tag.query.filter_by(name=name).first()
            if not tag:
                tag = Tag(name=name)
                db.session.add(tag)
            article.tags.append(tag)

class UserService:
    @staticmethod
    def create(email, username, password):
        """Create a new user."""
        user = User(email=email, username=username)
        user.set_password(password)

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

        # Send welcome email
        from app.tasks import send_welcome_email
        send_welcome_email.delay(user.id)

        return user

    @staticmethod
    def authenticate(email, password):
        """Authenticate user."""
        user = User.query.filter_by(email=email).first()

        if user and user.check_password(password) and user.is_active:
            return user

        return None
```

### Pattern 5: Decorators and Middleware

```python
# app/decorators.py
from functools import wraps
from flask import request, jsonify, g
from flask_jwt_extended import verify_jwt_in_request, current_user

def admin_required(f):
    """Require admin role."""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        verify_jwt_in_request()
        if not current_user.is_admin:
            return jsonify({'error': 'Admin access required'}), 403
        return f(*args, **kwargs)
    return decorated_function

def rate_limit(limit=100, period=60):
    """Rate limiting decorator."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            from app import redis_client

            key = f'rate_limit:{request.remote_addr}:{f.__name__}'
            current = redis_client.get(key)

            if current and int(current) >= limit:
                return jsonify({
                    'error': 'Rate limit exceeded',
                    'retry_after': redis_client.ttl(key)
                }), 429

            pipe = redis_client.pipeline()
            pipe.incr(key)
            pipe.expire(key, period)
            pipe.execute()

            return f(*args, **kwargs)
        return decorated_function
    return decorator

def validate_json(schema_class):
    """Validate request JSON with schema."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            schema = schema_class()
            try:
                g.validated_data = schema.load(request.json or {})
            except Exception as e:
                return jsonify({'error': 'Validation failed', 'details': e.messages}), 400
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# Usage
@api_v1.route('/admin/users', methods=['GET'])
@admin_required
def list_all_users():
    users = User.query.all()
    return jsonify(users_schema.dump(users))

@api_v1.route('/articles', methods=['POST'])
@jwt_required()
@validate_json(ArticleCreateSchema)
@rate_limit(limit=10, period=60)
def create_article():
    data = g.validated_data
    # Create article
```

### Pattern 6: Testing

```python
# tests/conftest.py
import pytest
from app import create_app, db
from app.models import User

@pytest.fixture
def app():
    app = create_app('testing')

    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def auth_client(client, user):
    from flask_jwt_extended import create_access_token

    with client.application.app_context():
        token = create_access_token(identity=user)

    client.environ_base['HTTP_AUTHORIZATION'] = f'Bearer {token}'
    return client

@pytest.fixture
def user(app):
    with app.app_context():
        user = User(email='test@example.com', username='testuser')
        user.set_password('password123')
        db.session.add(user)
        db.session.commit()
        return user

# tests/test_articles.py
def test_get_articles(client):
    response = client.get('/api/v1/articles')
    assert response.status_code == 200
    assert 'articles' in response.json

def test_create_article(auth_client):
    response = auth_client.post('/api/v1/articles', json={
        'title': 'Test Article',
        'body': 'This is test content.'
    })
    assert response.status_code == 201
    assert response.json['title'] == 'Test Article'

def test_create_article_unauthorized(client):
    response = client.post('/api/v1/articles', json={
        'title': 'Test Article',
        'body': 'This is test content.'
    })
    assert response.status_code == 401

def test_update_article_forbidden(auth_client, app):
    # Create article by different user
    with app.app_context():
        other_user = User(email='other@example.com', username='other')
        other_user.set_password('password')
        db.session.add(other_user)

        article = Article(
            title='Other Article',
            slug='other-article',
            body='Content',
            author=other_user
        )
        db.session.add(article)
        db.session.commit()
        slug = article.slug

    response = auth_client.put(f'/api/v1/articles/{slug}', json={
        'title': 'Updated Title'
    })
    assert response.status_code == 403
```

## Best Practices

### 1. Use Application Factory

```python
# Enables testing and multiple app instances
app = create_app('testing')
```

### 2. Organize with Blueprints

```python
# Modular application structure
from flask import Blueprint
api = Blueprint('api', __name__)
```

### 3. Use Proper HTTP Status Codes

```python
return jsonify(data), 201  # Created
return '', 204             # No Content
return jsonify(error), 400 # Bad Request
```

## Common Pitfalls

- **Circular Imports**: Use application factory pattern
- **No Request Context**: Use `with app.app_context()`
- **Blocking I/O**: Use Celery for long-running tasks
- **Missing Validation**: Always validate input data
- **SQL Injection**: Use SQLAlchemy ORM properly
- **Not Closing Sessions**: Use `db.session.remove()`

## Resources

- Flask Documentation
- Flask-RESTful Documentation
- Miguel Grinberg's Flask Tutorials
- Flask Mega-Tutorial