Flask API Development

Overview

Create efficient Flask APIs with blueprints for modular organization, SQLAlchemy for ORM, JWT authentication, comprehensive error handling, and proper request validation following REST principles.

When to Use

• Building RESTful APIs with Flask
• Creating microservices with minimal overhead
• Implementing lightweight authentication systems
• Designing API endpoints with proper validation
• Integrating with relational databases
• Building request/response handling systems

Instructions

1. Flask Application Setup

python
1# app.py
2from flask import Flask, request, jsonify
3from flask_cors import CORS
4from flask_sqlalchemy import SQLAlchemy
5from flask_jwt_extended import JWTManager
6import os
7
8app = Flask(__name__)
9app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'sqlite:///app.db')
10app.config['JWT_SECRET_KEY'] = os.getenv('JWT_SECRET_KEY', 'dev-secret')
11app.config['JSON_SORT_KEYS'] = False
12
13db = SQLAlchemy(app)
14jwt = JWTManager(app)
15CORS(app)
16
17# Request ID middleware
18@app.before_request
19def assign_request_id():
20    import uuid
21    request.request_id = str(uuid.uuid4())
22
23# Error handlers
24@app.errorhandler(400)
25def bad_request(error):
26    return jsonify({
27        'error': 'Bad Request',
28        'message': str(error),
29        'request_id': request.request_id
30    }), 400
31
32@app.errorhandler(404)
33def not_found(error):
34    return jsonify({
35        'error': 'Not Found',
36        'message': 'Resource does not exist',
37        'request_id': request.request_id
38    }), 404
39
40@app.errorhandler(500)
41def internal_error(error):
42    db.session.rollback()
43    return jsonify({
44        'error': 'Internal Server Error',
45        'request_id': request.request_id
46    }), 500
47
48if __name__ == '__main__':
49    app.run(debug=os.getenv('ENV') != 'production')

2. Database Models with SQLAlchemy

python
1# models.py
2from datetime import datetime
3from flask_sqlalchemy import SQLAlchemy
4from sqlalchemy.dialects.postgresql import UUID
5import uuid
6
7db = SQLAlchemy()
8
9class User(db.Model):
10    __tablename__ = 'users'
11
12    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
13    email = db.Column(db.String(255), unique=True, nullable=False, index=True)
14    password_hash = db.Column(db.String(255), nullable=False)
15    first_name = db.Column(db.String(100))
16    last_name = db.Column(db.String(100))
17    role = db.Column(db.String(20), default='user', index=True)
18    is_active = db.Column(db.Boolean, default=True)
19    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
20    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
21
22    # Relationships
23    posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
24
25    def __repr__(self):
26        return f'<User {self.email}>'
27
28    def set_password(self, password):
29        from werkzeug.security import generate_password_hash
30        self.password_hash = generate_password_hash(password)
31
32    def verify_password(self, password):
33        from werkzeug.security import check_password_hash
34        return check_password_hash(self.password_hash, password)
35
36    def to_dict(self):
37        return {
38            'id': str(self.id),
39            'email': self.email,
40            'first_name': self.first_name,
41            'last_name': self.last_name,
42            'role': self.role,
43            'created_at': self.created_at.isoformat()
44        }
45
46class Post(db.Model):
47    __tablename__ = 'posts'
48
49    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
50    title = db.Column(db.String(255), nullable=False, index=True)
51    content = db.Column(db.Text, nullable=False)
52    published = db.Column(db.Boolean, default=False)
53    user_id = db.Column(UUID(as_uuid=True), db.ForeignKey('users.id'), nullable=False)
54    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
55    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
56
57    def to_dict(self):
58        return {
59            'id': str(self.id),
60            'title': self.title,
61            'content': self.content,
62            'published': self.published,
63            'author_id': str(self.user_id),
64            'created_at': self.created_at.isoformat()
65        }

3. Authentication and JWT

python
1# auth.py
2from flask import request, jsonify
3from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity
4from functools import wraps
5from models import User, db
6
7def authenticate_user(email, password):
8    user = User.query.filter_by(email=email).first()
9    if user and user.verify_password(password):
10        return user
11    return None
12
13def login_required(f):
14    @wraps(f)
15    @jwt_required()
16    def decorated_function(*args, **kwargs):
17        identity = get_jwt_identity()
18        user = User.query.get(identity)
19        if not user or not user.is_active:
20            return jsonify({'error': 'User not found or inactive'}), 401
21        request.current_user = user
22        return f(*args, **kwargs)
23    return decorated_function
24
25def admin_required(f):
26    @wraps(f)
27    @login_required
28    def decorated_function(*args, **kwargs):
29        if request.current_user.role != 'admin':
30            return jsonify({'error': 'Admin access required'}), 403
31        return f(*args, **kwargs)
32    return decorated_function
33
34# routes/auth.py
35from flask import Blueprint, request, jsonify
36from auth import authenticate_user, login_required
37from models import User, db
38from flask_jwt_extended import create_access_token
39
40auth_bp = Blueprint('auth', __name__, url_prefix='/api/auth')
41
42@auth_bp.route('/login', methods=['POST'])
43def login():
44    data = request.get_json()
45    if not data or not data.get('email') or not data.get('password'):
46        return jsonify({'error': 'Missing credentials'}), 400
47
48    user = authenticate_user(data['email'], data['password'])
49    if not user:
50        return jsonify({'error': 'Invalid credentials'}), 401
51
52    access_token = create_access_token(identity=str(user.id))
53    return jsonify({
54        'access_token': access_token,
55        'user': user.to_dict()
56    }), 200
57
58@auth_bp.route('/register', methods=['POST'])
59def register():
60    data = request.get_json()
61    if User.query.filter_by(email=data['email']).first():
62        return jsonify({'error': 'Email already exists'}), 409
63
64    user = User(email=data['email'], first_name=data.get('first_name'))
65    user.set_password(data['password'])
66    db.session.add(user)
67    db.session.commit()
68
69    return jsonify({'user': user.to_dict()}), 201
70
71@auth_bp.route('/profile', methods=['GET'])
72@login_required
73def get_profile():
74    return jsonify({'user': request.current_user.to_dict()}), 200

4. Blueprints for Modular API Design

python
1# routes/users.py
2from flask import Blueprint, request, jsonify
3from auth import login_required, admin_required
4from models import User, db
5from sqlalchemy import or_
6
7users_bp = Blueprint('users', __name__, url_prefix='/api/users')
8
9@users_bp.route('', methods=['GET'])
10@login_required
11def list_users():
12    page = request.args.get('page', 1, type=int)
13    limit = request.args.get('limit', 20, type=int)
14    search = request.args.get('q', '', type=str)
15
16    query = User.query
17    if search:
18        query = query.filter(or_(
19            User.email.ilike(f'%{search}%'),
20            User.first_name.ilike(f'%{search}%')
21        ))
22
23    paginated = query.paginate(page=page, per_page=limit)
24    return jsonify({
25        'data': [user.to_dict() for user in paginated.items],
26        'pagination': {
27            'page': page,
28            'limit': limit,
29            'total': paginated.total,
30            'pages': paginated.pages
31        }
32    }), 200
33
34@users_bp.route('/<user_id>', methods=['GET'])
35@login_required
36def get_user(user_id):
37    user = User.query.get(user_id)
38    if not user:
39        return jsonify({'error': 'User not found'}), 404
40    return jsonify({'user': user.to_dict()}), 200
41
42@users_bp.route('/<user_id>', methods=['PATCH'])
43@login_required
44def update_user(user_id):
45    if str(request.current_user.id) != user_id:
46        return jsonify({'error': 'Unauthorized'}), 403
47
48    user = User.query.get(user_id)
49    if not user:
50        return jsonify({'error': 'User not found'}), 404
51
52    data = request.get_json()
53    if 'first_name' in data:
54        user.first_name = data['first_name']
55    if 'last_name' in data:
56        user.last_name = data['last_name']
57
58    db.session.commit()
59    return jsonify({'user': user.to_dict()}), 200
60
61@users_bp.route('/<user_id>', methods=['DELETE'])
62@admin_required
63def delete_user(user_id):
64    user = User.query.get(user_id)
65    if not user:
66        return jsonify({'error': 'User not found'}), 404
67
68    db.session.delete(user)
69    db.session.commit()
70    return '', 204

5. Request Validation

python
1# validators.py
2from flask import request, jsonify
3from functools import wraps
4
5def validate_json(*required_fields):
6    def decorator(f):
7        @wraps(f)
8        def decorated_function(*args, **kwargs):
9            if not request.is_json:
10                return jsonify({'error': 'Request body must be JSON'}), 400
11
12            data = request.get_json()
13            missing = [field for field in required_fields if field not in data]
14
15            if missing:
16                return jsonify({
17                    'error': 'Missing required fields',
18                    'missing_fields': missing
19                }), 400
20
21            return f(*args, **kwargs)
22        return decorated_function
23    return decorator
24
25def validate_email(email):
26    import re
27    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
28    return re.match(pattern, email) is not None
29
30# Usage
31@users_bp.route('', methods=['POST'])
32@validate_json('email', 'password', 'first_name')
33def create_user():
34    data = request.get_json()
35    if not validate_email(data['email']):
36        return jsonify({'error': 'Invalid email format'}), 400
37    # ... rest of logic

6. Application Factory and Configuration

python
1# config.py
2import os
3
4class Config:
5    SQLALCHEMY_TRACK_MODIFICATIONS = False
6    JSON_SORT_KEYS = False
7
8class DevelopmentConfig(Config):
9    DEBUG = True
10    TESTING = False
11    SQLALCHEMY_DATABASE_URI = 'sqlite:///app.db'
12
13class ProductionConfig(Config):
14    DEBUG = False
15    TESTING = False
16    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')
17    JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')
18
19class TestingConfig(Config):
20    TESTING = True
21    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
22
23# factory.py
24from flask import Flask
25from flask_sqlalchemy import SQLAlchemy
26from flask_jwt_extended import JWTManager
27
28def create_app(config_name='development'):
29    app = Flask(__name__)
30
31    if config_name == 'production':
32        from config import ProductionConfig
33        app.config.from_object(ProductionConfig)
34    else:
35        from config import DevelopmentConfig
36        app.config.from_object(DevelopmentConfig)
37
38    db = SQLAlchemy(app)
39    jwt = JWTManager(app)
40
41    # Register blueprints
42    from routes.auth import auth_bp
43    from routes.users import users_bp
44    app.register_blueprint(auth_bp)
45    app.register_blueprint(users_bp)
46
47    return app

Best Practices

✅ DO

• Use blueprints for modular organization
• Implement proper authentication with JWT
• Validate all user input
• Use SQLAlchemy ORM for database operations
• Implement comprehensive error handling
• Use pagination for collection endpoints
• Log errors and important events
• Return appropriate HTTP status codes
• Implement CORS properly
• Use environment variables for configuration

❌ DON'T

• Store secrets in code
• Use global variables for shared state
• Ignore database transactions
• Trust user input without validation
• Return stack traces in production
• Use mutable default arguments
• Forget to handle database connection errors
• Implement authentication in route handlers

Complete Example

python
1from flask import Flask, request, jsonify
2from flask_sqlalchemy import SQLAlchemy
3from flask_jwt_extended import JWTManager, create_access_token
4
5app = Flask(__name__)
6app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'
7db = SQLAlchemy(app)
8jwt = JWTManager(app)
9
10class User(db.Model):
11    id = db.Column(db.Integer, primary_key=True)
12    email = db.Column(db.String, unique=True)
13    password = db.Column(db.String)
14
15@app.route('/api/login', methods=['POST'])
16def login():
17    data = request.json
18    user = User.query.filter_by(email=data['email']).first()
19    if user:
20        token = create_access_token(identity=user.id)
21        return jsonify({'token': token}), 200
22    return jsonify({'error': 'Invalid'}), 401
23
24@app.route('/api/users', methods=['GET'])
25def get_users():
26    users = User.query.all()
27    return jsonify([{'id': u.id, 'email': u.email} for u in users]), 200
28
29if __name__ == '__main__':
30    app.run()