Transloadit
Publié : Dernière mise à jour :

Créer une API simple de traitement d’images avec Python et Flask

Kevin van Zonneveld
Kevin van Zonneveld
Cofondateur · Amsterdam, Pays-Bas ·

Une petite API d’images vous aide à comprendre comment les envois de fichiers, les décodeurs et les réponses HTTP s’articulent. Nous créons ici deux points de terminaison avec Flask et Pillow : l’un étire une image aux dimensions demandées, l’autre la convertit en PNG ou en JPEG. Tous deux utilisent le même parcours de validation et d’encodage.

Ce service de développement utilise l’interface de bouclage. Ce n’est pas un service public authentifié d’envoi de fichiers. Les limites de taille des fichiers, de pixels et de traitements simultanés réduisent l’utilisation des ressources ; elles ne constituent pas un bac à sable pour un décodeur natif.

Introduction aux API de traitement d’images

La requête contient un fichier multipart ainsi que les paramètres de l’opération. Le serveur vérifie le format décodé et les dimensions, effectue l’opération et renvoie les octets de l’image. Les noms de fichiers et les types MIME fournis par le navigateur ne prouvent pas qu’un fichier envoyé est sûr.

Configurer l’environnement de développement

Utilisez Python 3.12 ou une version ultérieure, Bash et cURL sous Linux. Vérifiez python3 --version et confirmez que curl --help all répertorie --fail-with-body avant de créer des fichiers. Dans un nouveau répertoire, créez requirements.txt :

Flask==3.1.3
Pillow==12.3.0
Flask-Limiter==4.1.1
gunicorn==26.2.0

Installez ensuite les paquets dans un environnement virtuel :

python3 -m venv .venv &&
source .venv/bin/activate &&
python -m pip install -r requirements.txt

Si la création de l’environnement ou l’installation échoue, corrigez l’erreur et répétez le bloc de configuration dans le même répertoire ; conservez requirements.txt et vos fichiers source. Les commandes s’arrêtent à l’étape qui échoue. L’environnement virtuel sépare les paquets de cet exemple de ceux d’un éventuel projet Python englobant.

Créer une application Flask de base

Placez cette application complète dans app.py. Elle comporte un seul bloc de démarrage, situé après toutes les routes et tous les gestionnaires :

import io
import re
import threading
import warnings

from flask import Flask, request, send_file
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from PIL import Image, ImageOps, UnidentifiedImageError
from werkzeug.exceptions import BadRequest, HTTPException

MAX_BYTES = 8 * 1024 * 1024
MAX_PIXELS = 12_000_000
MAX_SIDE = 2048
Image.MAX_IMAGE_PIXELS = MAX_PIXELS
warnings.simplefilter("error", Image.DecompressionBombWarning)

app = Flask(__name__)
# Leave room for Werkzeug's 64 KiB multipart parser chunks, including framing.
app.config.update(MAX_CONTENT_LENGTH=MAX_BYTES + 64 * 1024,
                  MAX_FORM_MEMORY_SIZE=128 * 1024, MAX_FORM_PARTS=8)
limiter = Limiter(get_remote_address, app=app,
                  default_limits=["60 per minute"], storage_uri="memory://")
slots = threading.BoundedSemaphore(2)


def dimension(name, default):
    values = request.form.getlist(name)
    text = values[0] if len(values) == 1 else default
    if len(values) > 1 or re.fullmatch(r"[0-9]{1,4}", text) is None:
        raise BadRequest()
    value = int(text)
    if not 1 <= value <= MAX_SIDE:
        raise BadRequest()
    return value


def load_image():
    files = request.files.getlist("file")
    if len(files) != 1 or len(request.files) != 1:
        raise BadRequest()
    data = files[0].read(MAX_BYTES + 1)
    if not data or len(data) > MAX_BYTES:
        raise BadRequest()
    with Image.open(io.BytesIO(data), formats=("JPEG", "PNG")) as source:
        # Pillow exposes some 16-bit PNGs as RGB/RGBA, so mode alone cannot enforce this limit.
        if source.format == "PNG" and (data[12:16] != b"IHDR" or data[24] > 8):
            raise BadRequest()
        if source.width * source.height > MAX_PIXELS or getattr(source, "n_frames", 1) != 1:
            raise BadRequest()
        source.load()
        oriented = ImageOps.exif_transpose(source)
        try:
            return oriented.convert("RGBA")
        finally:
            oriented.close()


def encode_image(image, target):
    image.info.clear()
    output = io.BytesIO()
    if target == "JPEG":
        background = Image.new("RGB", image.size, "white")
        try:
            background.paste(image, mask=image.getchannel("A"))
            background.save(output, format="JPEG", quality=85)
        finally:
            background.close()
    else:
        image.save(output, format="PNG")
    output.seek(0)
    response = send_file(output, mimetype=Image.MIME[target],
                         download_name="result." + ("jpg" if target == "JPEG" else "png"))
    response.headers["Cache-Control"] = "no-store"
    response.headers["X-Content-Type-Options"] = "nosniff"
    return response


def process_image(resize):
    if not slots.acquire(blocking=False):
        return {"error": "Image processor is busy."}, 503
    try:
        allowed = {"width", "height"} if resize else {"format"}
        if any(name not in allowed for name in request.form):
            raise BadRequest()
        formats = request.form.getlist("format")
        target = formats[0].upper() if formats else "PNG"
        if len(formats) > 1 or target not in ("PNG", "JPEG"):
            raise BadRequest()
        width = dimension("width", "100") if resize else None
        height = dimension("height", "100") if resize else None
        with load_image() as image:
            if resize:
                with image.resize((width, height), Image.Resampling.LANCZOS) as result:
                    return encode_image(result, "PNG")
            return encode_image(image, target)
    except (BadRequest, UnidentifiedImageError, OSError, ValueError,
            Image.DecompressionBombError, Image.DecompressionBombWarning):
        return {"error": "Provide one valid single-frame JPEG or PNG and supported parameters."}, 400
    finally:
        slots.release()


@app.post("/resize")
def resize_image():
    return process_image(resize=True)


@app.post("/convert")
def convert_image():
    return process_image(resize=False)


@app.errorhandler(HTTPException)
def http_error(error):
    messages = {413: "Upload is too large.", 429: "Too many requests."}
    return {"error": messages.get(error.code, "Request could not be processed.")}, error.code


@app.errorhandler(Exception)
def unexpected_error(error):
    # Avoid logging file content, headers or untrusted decoder messages.
    app.logger.error("Unexpected image processing failure: %s", type(error).__name__)
    return {"error": "Image processing failed."}, 500


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=False)

Intégrer Pillow pour le traitement d’images

load_image() ouvre uniquement les décodeurs JPEG et PNG. Cet exemple accepte les PNG comportant jusqu’à huit bits par canal et rejette les PNG à 16 bits plutôt que de modifier silencieusement leur plage de valeurs numériques. Il rejette les PNG animés et vérifie le nombre de pixels décodés avant de charger l’image. Il applique l’orientation EXIF et crée des pixels RGBA pour le parcours de traitement commun. La sortie supprime délibérément les métadonnées source, y compris les données GPS intégrées. Les profils de couleur ICC ne sont pas appliqués ; convertissez d’abord en sRGB les images d’entrée soumises à une gestion des couleurs.

La documentation des modes d’image de Pillow explique la représentation RGBA à huit bits et ses limites.

La limite de 8 MiB par fichier est distincte de la limite des requêtes multipart. Ni la taille du fichier compressé ni ses dimensions ne suffisent à elles seules à borner l’utilisation du processeur et de la mémoire par tous les décodeurs. Maintenez Pillow à jour et isolez les processus de traitement publics en appliquant des limites de ressources au niveau du système d’exploitation.

Implémenter le point de terminaison de redimensionnement d’images

Démarrez l’application avec python app.py. Laissez-la tourner et ouvrez un autre terminal dans le même répertoire pour utiliser cURL. Placez-y un fichier JPEG local nommé photo.jpg, puis redimensionnez-le :

curl --fail-with-body -F "file=@photo.jpg" -F "width=300" -F "height=200" \
  http://127.0.0.1:5000/resize --output resized.png

En cas de succès, cette commande renvoie une image PNG de 300 × 200 pixels. Cette opération étire l’image ; elle ne conserve pas son rapport largeur/hauteur. Chaque côté de l’image de sortie doit mesurer entre 1 et 2048 pixels. Pour des miniatures conservant ce rapport, choisissez explicitement une stratégie d’ajustement ou de recadrage plutôt que de modifier silencieusement la géométrie demandée.

Ouvrez resized.png pour vérifier son contenu. cURL remplace le fichier de sortie nommé lors d’une nouvelle exécution réussie ; avec --fail-with-body, une erreur HTTP peut y écrire du JSON à la place. Vérifiez donc également le code de sortie de cURL.

Ajouter le point de terminaison de conversion du format d’image

/convert accepte PNG ou JPEG, sans distinction de casse. Le format PNG conserve la transparence. Le format JPEG ne peut pas représenter la transparence ; l’exemple compose donc les pixels transparents sur un fond blanc. Placez un fichier PNG à huit bits nommé transparent.png dans le même répertoire, puis convertissez-le :

curl --fail-with-body -F "file=@transparent.png" -F "format=JPEG" \
  http://127.0.0.1:5000/convert --output converted.jpg

Ouvrez converted.jpg : les zones transparentes devraient être blanches et les pixels partiellement transparents devraient se mélanger au blanc. Les deux points de terminaison renvoient directement les images ; l’application n’enregistre ni les fichiers envoyés ni les résultats sur le serveur. Arrêtez le serveur avec Ctrl+C lorsque vous avez terminé.

Tester l’API avec des exemples de requêtes

Placez ces tests dans test_app.py et exécutez python -m unittest -v. Le client de test de Flask sollicite le véritable analyseur de requêtes et l’encodeur Pillow sans démarrer de serveur :

import io
import random
import unittest
from PIL import Image
from app import app, limiter

class ImageApiTest(unittest.TestCase):
    def setUp(self):
        app.config.update(TESTING=True)
        limiter.enabled = False
        self.client = app.test_client()

    def upload(self, path, input_format="PNG", **fields):
        image = io.BytesIO()
        mode = "RGB" if input_format == "JPEG" else "RGBA"
        with Image.new(mode, (80, 40), (200, 40, 80) if mode == "RGB" else (200, 40, 80, 128)) as source:
            source.paste((20, 60, 220) if mode == "RGB" else (20, 60, 220, 255), (52, 0, 80, 40))
            if mode == "RGBA":
                source.paste((0, 150, 250, 0), (0, 0, 24, 40))
            source.save(image, input_format)
        image.seek(0)
        return self.client.post(path, data={"file": (image, "photo"), **fields})

    def assert_pixel(self, image, position, expected, tolerance=5):
        for actual, wanted in zip(image.getpixel(position), expected):
            self.assertLessEqual(abs(actual - wanted), tolerance)

    def test_resize(self):
        response = self.upload("/resize", width="12", height="8")
        self.assertEqual(response.status_code, 200)
        with Image.open(io.BytesIO(response.data)) as image:
            self.assertEqual((image.format, image.size), ("PNG", (12, 8)))
            self.assertEqual(image.getpixel((0, 0))[3], 0)
            self.assert_pixel(image, (11, 7), (20, 60, 220, 255))

    def test_alpha_to_jpeg(self):
        response = self.upload("/convert", format="JPEG")
        self.assertEqual(response.status_code, 200)
        with Image.open(io.BytesIO(response.data)) as image:
            self.assertEqual(image.format, "JPEG")
            self.assert_pixel(image, (8, 20), (255, 255, 255))
            self.assert_pixel(image, (38, 20), (227, 147, 167))
            self.assert_pixel(image, (70, 20), (20, 60, 220))

    def test_jpeg_input(self):
        response = self.upload("/resize", input_format="JPEG", width="10", height="5")
        self.assertEqual(response.status_code, 200)
        with Image.open(io.BytesIO(response.data)) as image:
            self.assertEqual((image.format, image.size), ("PNG", (10, 5)))
            self.assert_pixel(image, (0, 0), (200, 40, 80, 255))
            self.assert_pixel(image, (9, 4), (20, 60, 220, 255))

    def test_rejects_16_bit_png(self):
        data = io.BytesIO()
        with Image.new("I;16", (40, 20), 32768) as image:
            image.save(data, "PNG")
        data.seek(0)
        response = self.client.post("/convert", data={"file": (data, "gray.png")})
        self.assertEqual(response.status_code, 400)

    def test_multipart_file_larger_than_a_parser_chunk(self):
        data = io.BytesIO()
        pixels = random.Random(0).randbytes(512 * 512 * 3)
        with Image.frombytes("RGB", (512, 512), pixels) as image:
            image.save(data, "JPEG", quality=90)
        self.assertGreater(data.tell(), 64 * 1024)
        data.seek(0)
        response = self.client.post("/resize", data={"file": (data, "photo.jpg")})
        self.assertEqual(response.status_code, 200)
        with Image.open(io.BytesIO(response.data)) as image:
            self.assertEqual(image.size, (100, 100))

    def test_invalid_dimensions(self):
        for width in ("0", "-1", "2049", "NaN", "12px"):
            with self.subTest(width=width):
                self.assertEqual(self.upload("/resize", width=width).status_code, 400)

    def test_invalid_upload(self):
        response = self.client.post("/resize", data={"file": (io.BytesIO(b"invalid"), "x.png")})
        self.assertEqual(response.status_code, 400)

if __name__ == "__main__":
    unittest.main()

Gestion des erreurs et validation des entrées

Les entrées invalides prévues reçoivent une réponse générique 400 ; une requête HTTP trop volumineuse reçoit une réponse 413. Chaque point de terminaison autorise 60 requêtes par minute et par adresse IP cliente. Les limites de débit renvoient une réponse 429, et une requête qui atteint l’application alors que les deux places de traitement sont occupées reçoit une réponse 503. Un serveur HTTP peut néanmoins mettre les connexions en attente avant qu’elles n’atteignent l’application. Les erreurs inattendues renvoient une réponse 500 expurgée des détails sensibles, jamais une exception du décodeur ni une trace de pile. Un fichier de plus de 8 MiB contenu dans une requête respectant la limite HTTP reçoit une réponse 400 ; les limites d’utilisation des ressources de Flask peuvent produire une réponse 413 plus tôt pour la requête, les champs de formulaire ou le nombre de parties.

Exécuter l’API locale avec Gunicorn

Sous Linux, un seul processus de travail Gunicorn maintient les limites en mémoire de cette démonstration dans un seul processus :

gunicorn --bind 127.0.0.1:5000 --workers 1 --threads 2 --timeout 30 \
  --no-control-socket app:app

Conservez cette écoute sur l’interface de bouclage. Configurez séparément un point d’entrée HTTPS authentifié et l’isolation des processus de travail avant d’accepter des envois de fichiers distants. Le socket de gestion facultatif est désactivé afin que cette démonstration ne partage pas le chemin du socket par défaut de Gunicorn avec une autre instance locale. Ctrl+C arrête le serveur ; l’arrêt des processus de travail peut prendre jusqu’à 30 secondes, soit le délai de grâce par défaut.

Avant d’ajouter des processus de travail ou des instances, configurez un stockage partagé des limites de débit à l’aide de la documentation du stockage de Flask-Limiter. Les compteurs en mémoire sont réinitialisés au redémarrage et ne sont pas partagés entre les processus. Un délai d’expiration du serveur n’est pas un mécanisme d’annulation par opération pour Pillow.

Conclusion et prochaines étapes

Complétez les tests pour couvrir les formats et la géométrie dont vous avez besoin, puis mesurez l’utilisation des ressources avec des fichiers représentatifs. Pour un flux de travail géré, explorez le service de traitement d’images de Transloadit plutôt que d’exploiter votre propre parc de décodeurs.

#python#flask#image-processing#api-development#handling-uploads-service

👩‍💻 Rejoignez plus de 20 000 développeurs

Inscrivez-vous à notre newsletter mensuelle  (English) pour recevoir des liens directs vers 3 contenus techniques exclusifs — et 2 nouveautés produit. Ni plus, ni moins.

Envoi et encodage de fichiers. En toute simplicité.

Transloadit simplifie la gestion des fichiers pour les développeurs et bénéficie de la confiance de marques comme Coursera et The New York Times. Nous sommes reconnus pour notre API fiable, notre support de premier ordre et notre fort engagement envers l’open source, avec des projets comme Uppy et tus qui font référence dans le traitement de fichiers.

S’inscrireRéserver une démo

Aucune carte bancaire requise

Résiliable à tout moment