Flask en Production: Créer des APIs Robustes et Maintenables

Flask en Production: Créer des APIs Robustes et Maintenables
Flask est léger, mais peut-il rivaliser avec Django en production? Découvrez comment construire une API Flask enterprise.

Structure Recommended
flask_api/
├── app/
│ ├── __init__.py # Factory pattern
│ ├── models.py # SQLAlchemy models
│ ├── config.py # Configuration
│ ├── utils.py # Utilitaires
│ │
│ ├── api/ # API Blueprint
│ │ ├── __init__.py
│ │ ├── users/
│ │ │ ├── routes.py
│ │ │ ├── schemas.py # Marshmallow
│ │ │ └── models.py
│ │ └── posts/
│ │
│ ├── services/ # Logique métier
│ │ ├── user_service.py
│ │ └── post_service.py
│ │
│ └── middleware/
│ ├── auth.py
│ └── errors.py
│
├── tests/
│ ├── test_users.py
│ └── conftest.py # Fixtures
│
├── migrations/
├── .env
├── requirements.txt
├── wsgi.py
└── README.md

Application Factory Pattern

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

db = SQLAlchemy()
migrate = Migrate()

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

# Configuration
if config_name == 'production':
from app.config import ProductionConfig
app.config.from_object(ProductionConfig)
else:
from app.config import DevelopmentConfig
app.config.from_object(DevelopmentConfig)

# Extensions
db.init_app(app)
migrate.init_app(app, db)
CORS(app)

# Blueprints
from app.api.users import users_bp
from app.api.posts import posts_bp

app.register_blueprint(users_bp, url_prefix='/api/v1/users')
app.register_blueprint(posts_bp, url_prefix='/api/v1/posts')

# Error handlers
@app.errorhandler(404)
def not_found(e):
return {'error': 'Not found'}, 404

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

return app

# wsgi.py
from app import create_app

app = create_app()

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

Models avec SQLAlchemy

# app/models.py
from app import db
from datetime import datetime

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)
name = db.Column(db.String(120), nullable=False)
password_hash = db.Column(db.String(255))
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)

posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')

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

def set_password(self, password):
from werkzeug.security import generate_password_hash
self.password_hash = generate_password_hash(password)

def check_password(self, password):
from werkzeug.security import check_password_hash
return check_password_hash(self.password_hash, password)

class Post(db.Model):
__tablename__ = 'posts'

id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(255), nullable=False, index=True)
content = db.Column(db.Text)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)

def __repr__(self):
return f'<Post {self.title}>'

Validation avec Marshmallow

# app/api/users/schemas.py
from marshmallow import Schema, fields, validate, post_load

class UserSchema(Schema):
id = fields.Int(dump_only=True)
email = fields.Email(required=True, validate=validate.Length(min=5))
name = fields.Str(required=True, validate=validate.Length(min=3, max=255))
created_at = fields.DateTime(dump_only=True)

class Meta:
strict = True

class UserDetailSchema(UserSchema):
posts = fields.Nested('PostSchema', many=True, dump_only=True)

class PostSchema(Schema):
id = fields.Int(dump_only=True)
title = fields.Str(required=True, validate=validate.Length(min=3))
content = fields.Str()
author = fields.Nested(UserSchema, dump_only=True)
created_at = fields.DateTime(dump_only=True)

Routes et Endpoints

# app/api/users/routes.py
from flask import Blueprint, request, jsonify
from app import db
from app.models import User
from app.services.user_service import UserService
from app.api.users.schemas import UserSchema, UserDetailSchema
from app.middleware.auth import login_required

users_bp = Blueprint('users', __name__)
user_schema = UserSchema()
users_schema = UserSchema(many=True)
user_detail_schema = UserDetailSchema()

@users_bp.route('', methods=['GET'])
def get_users():
page = request.args.get('page', 1, type=int)
users = User.query.paginate(page=page, per_page=20)

return jsonify({
'data': users_schema.dump(users.items),
'total': users.total,
'pages': users.pages
})

@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
user = User.query.get_or_404(user_id)
return jsonify(user_detail_schema.dump(user))

@users_bp.route('', methods=['POST'])
def create_user():
data = request.get_json()

# Validation
errors = user_schema.validate(data)
if errors:
return jsonify(errors), 400

# Créer
try:
user = UserService.create_user(
email=data['email'],
name=data['name'],
password=data.get('password')
)
return jsonify(user_schema.dump(user)), 201
except ValueError as e:
return jsonify({'error': str(e)}), 409

@users_bp.route('/<int:user_id>', methods=['PUT'])
@login_required
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()

if 'name' in data:
user.name = data['name']

db.session.commit()
return jsonify(user_schema.dump(user))

@users_bp.route('/<int:user_id>', methods=['DELETE'])
@login_required
def delete_user(user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204

Services (Logique Métier)

# app/services/user_service.py
from app import db
from app.models import User

class UserService:
@staticmethod
def create_user(email, name, password):
if User.query.filter_by(email=email).first():
raise ValueError('Email déjà utilisé')

user = User(email=email, name=name)
if password:
user.set_password(password)

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

@staticmethod
def get_user_with_posts(user_id):
return User.query.options(
db.joinedload(User.posts)
).get_or_404(user_id)

Authentification JWT

# app/middleware/auth.py
from flask import request, jsonify
from functools import wraps
import jwt
import os

def login_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
token = request.headers.get('Authorization', '').replace('Bearer ', '')

if not token:
return jsonify({'error': 'Token manquant'}), 401

try:
payload = jwt.decode(token, os.getenv('JWT_SECRET'), algorithms=['HS256'])
request.user_id = payload['user_id']
except jwt.InvalidTokenError:
return jsonify({'error': 'Token invalide'}), 401

return f(*args, **kwargs)

return decorated_function
Testing

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

# tests/test_users.py
def test_create_user(client):
response = client.post('/api/v1/users', json={
'email': 'test@example.com',
'name': 'Test User',
'password': 'secure123'
})

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

def test_get_users(client):
response = client.get('/api/v1/users')
assert response.status_code == 200
assert 'data' in response.json

Configuration Production

# app/config.py
import os

class Config:
SQLALCHEMY_TRACK_MODIFICATIONS = False
JSON_SORT_KEYS = False

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

class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10,
'pool_recycle': 3600,
'pool_pre_ping': True,
}

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

Déploiement avec Gunicorn

# requirements.txt
Flask==3.0.0
Flask-SQLAlchemy==3.1.1
Flask-Migrate==4.0.5
Flask-CORS==4.0.0
Marshmallow==3.20.1
python-dotenv==1.0.0
gunicorn==21.2.0
psycopg2-binary==2.9.9
PyJWT==2.8.1

# Déployer
gunicorn -w 4 -b 0.0.0.0:5000 wsgi:app

Questions pour la Communauté

Flask vs Django: quand choisir Flask?
Utilisez-vous des extensions supplémentaires?
Comment gérez-vous la scalabilité avec Flask?
Avez-vous des patterns Flask éprouvés à partager?
Partagez vos retours d'expérience!