Einfache Bildverarbeitungs-API mit Python und Flask erstellen

Kevin van Zonneveld

Mitgründer · Amsterdam, Niederlande ·

Eine kleine Bild-API zeigt Ihnen, wie Uploads, Decoder und HTTP-Antworten zusammenspielen. Hier bauen wir zwei Endpunkte mit Flask und Pillow: Der eine streckt ein Bild auf eine angeforderte Größe, der andere konvertiert es nach PNG oder JPEG. Beide nutzen denselben Pfad für Validierung und Encoding.

Dies ist ein Entwicklungsdienst auf dem Loopback-Interface, kein authentifizierter öffentlicher Upload-Dienst. Limits für Dateigröße, Pixel und Parallelität senken den Ressourcenverbrauch; sie machen aus einem nativen Decoder aber keine Sandbox.

Einführung in Bildverarbeitungs-APIs

Die Anfrage enthält eine Multipart-Datei sowie Parameter für den Vorgang. Der Server prüft Format und Abmessungen nach dem Decodieren, führt den Vorgang aus und gibt Bild-Bytes zurück. Dateinamen und vom Browser gelieferte MIME-Typen sind kein Beleg dafür, dass ein Upload sicher ist.

Die Entwicklungsumgebung einrichten

Verwenden Sie Python 3.12 oder neuer. Erstellen Sie in einem neuen Verzeichnis requirements.txt:

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

Installieren Sie anschließend in eine virtuelle Umgebung:

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

Unter Windows aktivieren Sie die Umgebung mit .venv\Scripts\activate. Halten Sie die aufgelösten Abhängigkeitsversionen in Ihrem Deployment-Lockfile fest und prüfen Sie Updates, bevor Sie einen Dienst für Uploads öffnen.

Eine einfache Flask-Anwendung erstellen

Legen Sie diese vollständige Anwendung in app.py ab. Es gibt einen Startblock, und zwar nach allen Routen und Handlern:

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:
        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)

Pillow für die Bildverarbeitung einbinden

load_image() öffnet ausschließlich JPEG- und PNG-Decoder. Die Funktion lehnt animierte PNGs ab und prüft die decodierte Pixelanzahl, bevor das Bild geladen wird. Sie wendet die EXIF-Ausrichtung an und erzeugt RGBA-Pixel für den gemeinsamen Verarbeitungspfad. Die Ausgabe verwirft die Metadaten der Quelle bewusst, einschließlich eingebetteter GPS-Daten.

Das Dateilimit von 8 MiB ist unabhängig vom Limit für Multipart-Anfragen. Weder die komprimierte Dateigröße noch die Abmessungen allein begrenzen den CPU- und Speicherverbrauch jedes Decoders. Halten Sie Pillow aktuell und isolieren Sie öffentliche Verarbeitungs-Worker mit Ressourcenlimits des Betriebssystems.

Endpunkt für die Größenänderung von Bildern implementieren

Starten Sie die Anwendung mit python app.py und ändern Sie dann die Größe eines lokalen Bildes:

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

Das Ergebnis ist immer ein PNG mit 300 mal 200 Pixeln. Dieser Vorgang streckt das Bild; das Seitenverhältnis bleibt dabei nicht erhalten. Jede Seite der Ausgabe muss zwischen 1 und 2048 Pixeln liegen. Für Thumbnails mit erhaltenem Seitenverhältnis wählen Sie explizit eine Fit- oder Crop-Strategie, statt die angeforderte Geometrie stillschweigend zu ändern.

Endpunkt für die Formatkonvertierung von Bildern hinzufügen

/convert akzeptiert PNG oder JPEG, unabhängig von Groß- und Kleinschreibung. PNG behält Transparenz bei. JPEG kann Transparenz nicht abbilden, daher kombiniert das Beispiel transparente Pixel mit einem weißen Hintergrund:

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

Die API mit Beispielanfragen testen

Legen Sie diese Tests in test_app.py ab und führen Sie python -m unittest -v aus. Der Test-Client von Flask nutzt den tatsächlichen Request-Parser und den Pillow-Encoder, ohne einen Server zu starten:

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"
        Image.new(mode, (40, 20), 0).save(image, input_format)
        image.seek(0)
        return self.client.post(path, data={"file": (image, "photo"), **fields})

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

    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.assertEqual(image.getpixel((0, 0)), (255, 255, 255))

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

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

Fehlerbehandlung und Eingabevalidierung

Erwartbar ungültige Eingaben erhalten eine generische 400-Antwort; eine zu große HTTP-Anfrage erhält 413. Rate Limits geben 429 zurück, und die Parallelitätsgrenze pro Prozess gibt 503 zurück, ohne weitere Bildaufgaben in die Queue zu stellen. Unerwartete Fehler geben eine bereinigte 500-Antwort zurück, niemals eine Decoder-Ausnahme oder einen Stacktrace.

Die API auf einem Server bereitstellen

Auf einem Unix-Server hält ein einzelner Gunicorn-Worker die In-Memory-Limits dieser Demo in einem einzigen Prozess:

gunicorn --bind 127.0.0.1:5000 --workers 1 --threads 2 --timeout 30 app:app

Ein echtes Deployment benötigt zusätzlich TLS an einem Reverse Proxy, Authentifizierung und Autorisierung, Zeitlimits für Anfragen, Body-Limits am Proxy, Monitoring und isolierte Worker. CORS ist keine Authentifizierung. Vertrauen Sie weitergeleiteten IP-Headern erst, wenn Ihre Proxy-Topologie verhindert, dass Clients sie selbst setzen.

Bevor Sie Worker oder Instanzen hinzufügen, konfigurieren Sie mithilfe der Flask-Limiter-Dokumentation zum Storage einen gemeinsamen Rate-Limit-Store. Speicherbasierte Zähler werden bei einem Neustart zurückgesetzt und nicht zwischen Prozessen geteilt. Ein Server-Timeout ist kein Mechanismus, um einzelne Vorgänge in Pillow abzubrechen.

Fazit und nächste Schritte

Diese API bündelt Upload-Validierung und Encoding an einer Stelle und stellt dabei zwei Vorgänge bereit. Erweitern Sie sie um Tests für die Formate und Geometrien, die Sie benötigen, und messen Sie anschließend den Ressourcenverbrauch mit repräsentativen Dateien. Für einen verwalteten Workflow werfen Sie einen Blick auf den Dienst für Bildverarbeitung von Transloadit, statt eine eigene Decoder-Flotte zu betreiben.