Einfache Bildverarbeitungs-API mit Python und Flask erstellen

Kevin van Zonneveld

Mitgründer · Amsterdam, Niederlande ·

Mit einer kleinen Bild-API lernen Sie, wie Uploads, Decoder und HTTP-Antworten zusammenspielen. Hier erstellen wir zwei Endpunkte mit Flask und Pillow: Einer streckt ein Bild auf die gewünschte Größe, der andere konvertiert es in PNG oder JPEG. Beide nutzen denselben Validierungs- und Encoding-Pfad.

Dies ist ein Entwicklungsdienst auf der Loopback-Schnittstelle, kein authentifizierter öffentlicher Upload-Dienst. Grenzen für Dateigröße, Pixelzahl und Parallelität reduzieren den Ressourcenverbrauch; sie machen einen nativen Decoder nicht zur Sandbox.

Einführung in Bildverarbeitungs-APIs

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

Entwicklungsumgebung einrichten

Verwenden Sie Python 3.12 oder neuer, Bash und cURL unter Linux. Prüfen Sie python3 --version und bestätigen Sie, dass curl --help all den Eintrag --fail-with-body auflistet, bevor Sie Dateien erstellen. 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 die Pakete in einer virtuellen Umgebung:

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

Falls das Erstellen der Umgebung oder die Installation fehlschlägt, beheben Sie den Fehler und wiederholen Sie den Einrichtungsblock im selben Verzeichnis. Behalten Sie requirements.txt und Ihre Quelldateien bei. Die Befehle stoppen beim fehlgeschlagenen Schritt. Die virtuelle Umgebung hält die Pakete dieses Beispiels von denen eines übergeordneten Python-Projekts getrennt.

Eine einfache Flask-Anwendung erstellen

Speichern Sie diese vollständige Anwendung in app.py. Es gibt genau einen Startblock, 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:
        # 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)

Pillow für die Bildverarbeitung integrieren

load_image() öffnet nur JPEG- und PNG-Decoder. Dieses Beispiel akzeptiert PNGs mit bis zu acht Bit pro Kanal und lehnt 16-Bit-PNGs ab, statt ihren Wertebereich stillschweigend zu ändern. Es lehnt animierte PNGs ab und prüft die decodierte Pixelzahl, bevor es das Bild lädt. Es wendet die EXIF-Ausrichtung an und erzeugt RGBA-Pixel für den gemeinsamen Verarbeitungspfad. Die Ausgabe verwirft bewusst die Quellmetadaten, einschließlich eingebetteter GPS-Daten. ICC-Farbprofile werden nicht angewendet; konvertieren Sie Eingaben mit Farbmanagement zuerst in sRGB.

Die Dokumentation zu Bildmodi von Pillow erläutert die RGBA-Darstellung mit acht Bit und ihre Grenzen.

Das Dateilimit von 8 MiB gilt 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 Worker für die öffentliche Verarbeitung mit Ressourcenlimits auf Betriebssystemebene.

Endpunkt zur Größenänderung von Bildern implementieren

Starten Sie die Anwendung mit python app.py. Lassen Sie sie laufen und öffnen Sie für cURL ein weiteres Terminal im selben Verzeichnis. Legen Sie dort eine lokale JPEG-Datei namens photo.jpg ab und ändern Sie dann ihre Größe:

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

Bei Erfolg gibt dieser Befehl ein PNG mit 300 mal 200 Pixeln zurück. Der Vorgang streckt das Bild; das Seitenverhältnis bleibt nicht erhalten. Jede Ausgabeseite muss zwischen 1 und 2048 Pixeln lang sein. Wählen Sie für Thumbnails mit unverändertem Seitenverhältnis explizit eine Einpassungs- oder Zuschnittstrategie, statt die angeforderten Abmessungen stillschweigend zu ändern.

Öffnen Sie resized.png, um den Inhalt zu prüfen. Bei einer erfolgreichen erneuten Ausführung ersetzt cURL die benannte Ausgabedatei. Mit --fail-with-body kann bei einem HTTP-Fehler stattdessen JSON in diese Datei geschrieben werden. Prüfen Sie daher auch den Exit-Status von cURL.

Endpunkt zur Bildformatkonvertierung hinzufügen

/convert akzeptiert PNG oder JPEG, unabhängig von Groß- und Kleinschreibung. PNG erhält Transparenz. JPEG kann Transparenz nicht darstellen, daher legt das Beispiel transparente Pixel über einen weißen Hintergrund. Legen Sie ein Acht-Bit-PNG namens transparent.png im selben Verzeichnis ab und konvertieren Sie es dann:

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

Öffnen Sie converted.jpg: Transparente Bereiche sollten weiß sein, und teiltransparente Pixel sollten mit Weiß vermischt sein. Beide Endpunkte geben Bilder direkt zurück; die Anwendung speichert weder Uploads noch Ergebnisse auf dem Server. Stoppen Sie den Server mit Ctrl+C, wenn Sie fertig sind.

API mit Beispielanfragen testen

Speichern Sie diese Tests in test_app.py und führen Sie python -m unittest -v aus. Der Testclient von Flask testet den tatsächlichen Anfrageparser 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"
        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()

Fehlerbehandlung und Eingabevalidierung

Erwartete ungültige Eingaben erhalten eine generische Antwort mit Status 400; eine zu große HTTP-Anfrage erhält 413. Jeder Endpunkt erlaubt 60 Anfragen pro Minute und Client-IP. Bei überschrittenen Ratenlimits wird 429 zurückgegeben. Erreicht eine Anfrage die Anwendung, während beide Verarbeitungsplätze belegt sind, erhält sie 503. Ein HTTP-Server kann Verbindungen weiterhin in eine Warteschlange stellen, bevor sie die Anwendung erreichen. Unerwartete Fehler führen zu einer bereinigten Antwort mit Status 500, niemals zu einer Decoder-Ausnahme oder einem Stacktrace. Eine Datei über 8 MiB innerhalb des HTTP-Anfragelimits erhält 400; die Ressourcenlimits von Flask können bereits früher wegen der Anfrage, der Formularfelder oder der Anzahl der Teile zu 413 führen.

Lokale API mit Gunicorn ausführen

Unter Linux hält ein einzelner Gunicorn-Worker die In-Memory-Limits dieser Demo in einem Prozess:

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

Lassen Sie diesen Listener an die Loopback-Schnittstelle gebunden. Konfigurieren Sie einen authentifizierten HTTPS-Zugangspunkt und die Worker-Isolation separat, bevor Sie externe Uploads annehmen. Der optionale Verwaltungssocket ist deaktiviert, damit diese Demo den Standard-Socketpfad von Gunicorn nicht mit einer anderen lokalen Instanz teilt. Ctrl+C stoppt den Server; das Herunterfahren der Worker kann bis zum Ablauf der standardmäßigen Wartefrist von 30 Sekunden dauern.

Bevor Sie Worker oder Instanzen hinzufügen, konfigurieren Sie einen gemeinsamen Speicher für Ratenlimits mithilfe der Dokumentation zum Speicher von Flask-Limiter. Zähler im Arbeitsspeicher werden beim Neustart zurückgesetzt und nicht prozessübergreifend geteilt. Ein Server-Timeout ist kein Mechanismus zum Abbrechen einzelner Pillow-Vorgänge.

Fazit und nächste Schritte

Erweitern Sie die Tests für Ihre benötigten Formate und Abmessungen und messen Sie anschließend den Ressourcenverbrauch mit repräsentativen Dateien. Für einen verwalteten Workflow entdecken Sie den Service zur Bildverarbeitung von Transloadit, statt eine eigene Decoder-Flotte zu betreiben.