Crear API simple de procesamiento de imágenes con Python y Flask

Kevin van Zonneveld

Cofundador · Ámsterdam, Países Bajos ·

Una pequeña API de imágenes te ayuda a aprender cómo se relacionan las subidas, los decodificadores y las respuestas HTTP. Aquí creamos dos endpoints con Flask y Pillow: uno estira una imagen hasta el tamaño solicitado y el otro la convierte a PNG o JPEG. Ambos usan la misma ruta de validación y codificación.

Este es un servicio de desarrollo en la interfaz de bucle local, no un servicio público de subidas con autenticación. Los límites de tamaño de archivo, píxeles y concurrencia reducen el uso de recursos; no convierten un decodificador nativo en un entorno aislado.

Introducción a las API de procesamiento de imágenes

La solicitud contiene un archivo multipart y los parámetros de la operación. El servidor comprueba el formato y las dimensiones decodificados, realiza la operación y devuelve los bytes de la imagen. Los nombres de archivo y los tipos MIME proporcionados por el navegador no demuestran que un archivo subido sea seguro.

Configurar el entorno de desarrollo

Usa Python 3.12 o posterior, Bash y cURL en Linux. Comprueba python3 --version y confirma que curl --help all incluye --fail-with-body antes de crear archivos. En un directorio nuevo, crea requirements.txt:

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

Luego instala los paquetes en un entorno virtual:

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

Si falla la creación del entorno o la instalación, resuelve ese error y repite el bloque de configuración en el mismo directorio; conserva requirements.txt y tus archivos de código fuente. Los comandos se detienen en el paso que falla. El entorno virtual mantiene los paquetes de este ejemplo separados de los de un proyecto Python que lo contenga.

Crear una aplicación básica con Flask

Coloca esta aplicación completa en app.py. Hay un único bloque de inicio, después de todas las rutas y los manejadores:

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)

Integrar Pillow para el procesamiento de imágenes

load_image() abre únicamente los decodificadores JPEG y PNG. Este ejemplo acepta archivos PNG de hasta ocho bits por canal y rechaza los PNG de 16 bits en lugar de cambiar su rango numérico sin avisar. Rechaza los PNG animados y comprueba la cantidad de píxeles decodificados antes de cargar la imagen. Aplica la orientación EXIF y crea píxeles RGBA para la ruta de procesamiento compartida. La salida descarta deliberadamente los metadatos de origen, incluidos los datos GPS incrustados. No aplica perfiles de color ICC; convierte primero a sRGB las imágenes de entrada con gestión de color.

La documentación de los modos de imagen de Pillow explica la representación RGBA de ocho bits y sus límites.

El límite de archivo de 8 MiB es independiente del límite de la solicitud multipart. Ni el tamaño del archivo comprimido ni las dimensiones por sí solos limitan el uso de CPU y memoria de todos los decodificadores. Mantén Pillow actualizado y aísla los procesos de trabajo que procesan solicitudes públicas mediante límites de recursos del sistema operativo.

Implementar el endpoint de redimensionamiento de imágenes

Inicia la aplicación con python app.py. Déjala en ejecución y abre otra terminal en el mismo directorio para cURL. Coloca allí un archivo JPEG local llamado photo.jpg y luego cambia su tamaño:

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

Si se ejecuta correctamente, este comando devuelve un PNG de 300 por 200. Esta operación estira la imagen; no conserva la relación de aspecto. Cada lado de la imagen de salida debe medir entre 1 y 2048 píxeles. Para miniaturas que conserven la relación de aspecto, elige explícitamente una política de ajuste o recorte en lugar de cambiar la geometría solicitada sin avisar.

Abre resized.png para comprobar su contenido. cURL reemplaza el archivo de salida indicado cuando vuelves a ejecutar el comando correctamente; con --fail-with-body, un error HTTP puede escribir JSON en ese archivo, así que comprueba también el estado de salida de cURL.

Añadir el endpoint de conversión de formatos de imagen

/convert acepta PNG o JPEG, sin distinguir mayúsculas de minúsculas. PNG conserva la transparencia. JPEG no puede representar transparencia, por lo que el ejemplo compone los píxeles transparentes sobre un fondo blanco. Coloca un PNG de ocho bits llamado transparent.png en el mismo directorio y luego conviértelo:

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

Abre converted.jpg: las áreas transparentes deberían ser blancas y los píxeles parcialmente transparentes deberían mezclarse con el blanco. Ambos endpoints devuelven imágenes directamente; la aplicación no guarda los archivos subidos ni los resultados en el servidor. Detén el servidor con Ctrl+C cuando termines.

Probar la API con solicitudes de ejemplo

Coloca estas pruebas en test_app.py y ejecuta python -m unittest -v. El cliente de pruebas de Flask ejercita el analizador de solicitudes y el codificador de Pillow reales sin iniciar un servidor:

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

Manejo de errores y validación de entradas

Las entradas no válidas previstas reciben una respuesta genérica 400; una solicitud HTTP que excede el tamaño permitido recibe 413. Cada endpoint permite 60 solicitudes por minuto por IP de cliente. Los límites de frecuencia devuelven 429, y una solicitud que llega a la aplicación mientras ambos cupos de procesamiento están ocupados recibe 503. Un servidor HTTP aún puede poner conexiones en cola antes de que lleguen a la aplicación. Los fallos inesperados devuelven una respuesta 500 sin información sensible, nunca una excepción del decodificador ni una traza de la pila. Un archivo de más de 8 MiB dentro del límite de la solicitud HTTP recibe 400; los límites de uso de recursos de Flask pueden producir un 413 antes debido a la solicitud, los campos del formulario o la cantidad de partes.

Ejecuta la API local con Gunicorn

En Linux, un único proceso de trabajo de Gunicorn mantiene los límites en memoria de este ejemplo en un solo proceso:

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

Mantén este servicio de escucha en la interfaz de bucle local. Configura por separado un punto de entrada HTTPS con autenticación y el aislamiento de los procesos de trabajo antes de aceptar subidas remotas. El socket de administración opcional está deshabilitado para que este ejemplo no comparta la ruta de socket predeterminada de Gunicorn con otra instancia local. Ctrl+C detiene el servidor; el cierre de los procesos de trabajo puede tardar hasta los 30 segundos del período de gracia predeterminado.

Antes de añadir procesos de trabajo o instancias, configura un almacén compartido para los límites de frecuencia mediante la documentación de almacenamiento de Flask-Limiter. Los contadores almacenados en memoria se restablecen al reiniciar y no se comparten entre procesos. Un tiempo de espera del servidor no es un mecanismo de cancelación por operación para Pillow.

Conclusión y próximos pasos

Amplía las pruebas para los formatos y la geometría que necesites y luego mide el uso de recursos con archivos representativos. Para un flujo de trabajo administrado, explora el servicio de procesamiento de imágenes de Transloadit en lugar de operar tu propio conjunto de decodificadores.