Crear una API sencilla de procesamiento de imágenes con Python y Flask

Kevin van Zonneveld

Cofundador · Ámsterdam, Países Bajos ·

Una API de imágenes pequeña te ayuda a entender cómo encajan las subidas, los decodificadores y las respuestas HTTP. Aquí construimos 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 loopback, no un servicio público de subidas con autenticación. Los límites de tamaño de archivo, de píxeles y de concurrencia reducen el uso de recursos; no convierten un decodificador nativo en un sandbox.

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

La solicitud contiene un archivo multipart más los parámetros de la operación. El servidor comprueba el formato y las dimensiones decodificados, ejecuta la operación y devuelve los bytes de la imagen. Los nombres de archivo y los tipos MIME que proporciona el navegador no son prueba de que una subida sea segura.

Configurar el entorno de desarrollo

Usa Python 3.12 o una versión más reciente. En un directorio nuevo, crea requirements.txt:

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

Después, instala en un entorno virtual:

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

En Windows, activa el entorno con .venv\Scripts\activate. Mantén las versiones de dependencias resueltas en el lockfile de tu despliegue y revisa las actualizaciones antes de exponer un servicio a subidas.

Crear una aplicación básica de Flask

Coloca esta aplicación completa en app.py. Hay un único bloque de arranque, 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:
        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() solo abre los decodificadores JPEG y PNG. Rechaza los PNG animados y comprueba el número 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 de GPS incrustados.

El límite de 8 MiB por archivo es independiente del límite de la solicitud multipart. Ni el tamaño del archivo comprimido ni las dimensiones por sí solos acotan el uso de CPU y de memoria de todos los decodificadores. Mantén Pillow actualizado y aísla los workers de procesamiento públicos con límites de recursos del sistema operativo.

Implementar el endpoint de redimensionado de imágenes

Inicia la aplicación con python app.py y luego redimensiona una imagen local:

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

El resultado siempre es un PNG de 300 por 200. Esta operación estira la imagen; no conserva la relación de aspecto. Cada lado de la salida debe estar entre 1 y 2048 píxeles. Para miniaturas que conserven la relación de aspecto, elige explícitamente una política de ajuste o de recorte en lugar de cambiar en silencio la geometría solicitada.

Agregar el endpoint de conversión de formato de imagen

/convert acepta PNG o JPEG, sin distinguir mayúsculas de minúsculas. PNG conserva la transparencia. JPEG no puede representar la transparencia, así que el ejemplo compone los píxeles transparentes sobre blanco:

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

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 real y el codificador de Pillow 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"
        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()

Manejo de errores y validación de entradas

Las entradas inválidas previstas reciben una respuesta 400 genérica; una solicitud HTTP demasiado grande recibe un 413. Los límites de tasa devuelven 429 y el tope de concurrencia por proceso devuelve 503 sin poner en cola más trabajo de imágenes. Los fallos inesperados devuelven una respuesta 500 saneada, nunca una excepción del decodificador ni una traza de pila.

Desplegar la API en un servidor

En un servidor Unix, un único worker de Gunicorn mantiene en un solo proceso los límites en memoria de esta demo:

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

Un despliegue real también necesita TLS en un proxy inverso, autenticación y autorización, plazos límite para las solicitudes, límites de tamaño del cuerpo en el proxy, monitoreo y workers aislados. CORS no es autenticación. No confíes en las cabeceras de IP reenviadas hasta que la topología de tu proxy impida que los clientes las proporcionen.

Antes de agregar workers o instancias, configura un almacén compartido de límites de tasa con la documentación de almacenamiento de Flask-Limiter. Los contadores 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

Esta API mantiene la validación de subidas y la codificación en un solo lugar mientras expone dos operaciones. Amplíala con 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 gestionado, explora el servicio de procesamiento de imágenes de Transloadit en lugar de mantener tu propia flota de decodificadores.