Criando uma API simples para processar imagens com Python e Flask

Kevin van Zonneveld

Cofundador · Amsterdã, Países Baixos ·

Uma pequena API de imagens ajuda você a entender como uploads, decodificadores e respostas HTTP se encaixam. Aqui criamos dois endpoints com Flask e Pillow: um estica a imagem para o tamanho solicitado e o outro a converte para PNG ou JPEG. Ambos usam o mesmo caminho de validação e codificação.

Este é um serviço de desenvolvimento em loopback, não um serviço público de upload autenticado. Limites de tamanho de arquivo, de pixels e de concorrência reduzem o uso de recursos, mas não transformam um decodificador nativo em uma sandbox.

Introdução às APIs de processamento de imagens

A requisição contém um arquivo multipart mais os parâmetros da operação. O servidor verifica o formato e as dimensões decodificados, executa a operação e retorna os bytes da imagem. Nomes de arquivo e tipos MIME informados pelo navegador não são evidência de que um upload é seguro.

Configurando o ambiente de desenvolvimento

Use Python 3.12 ou mais recente. Em um novo diretório, crie requirements.txt:

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

Depois, instale em um ambiente virtual:

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

No Windows, ative o ambiente com .venv\Scripts\activate. Mantenha as versões resolvidas das dependências no lockfile da sua implantação e revise as atualizações antes de expor um serviço a uploads.

Criando uma aplicação Flask básica

Coloque esta aplicação completa em app.py. Há um único bloco de inicialização, depois de todas as rotas e manipuladores:

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)

Integrando o Pillow para processamento de imagens

load_image() abre apenas os decodificadores de JPEG e PNG. Ele rejeita PNGs animados e verifica a contagem de pixels decodificados antes de carregar a imagem. Ele aplica a orientação EXIF e cria pixels RGBA para o caminho de processamento compartilhado. A saída descarta deliberadamente os metadados de origem, incluindo dados de GPS incorporados.

O limite de 8 MiB por arquivo é separado do limite da requisição multipart. Nem o tamanho do arquivo comprimido nem as dimensões, isoladamente, limitam o uso de CPU e memória de todo decodificador. Mantenha o Pillow atualizado e isole os workers públicos de processamento com limites de recursos do sistema operacional.

Implementando o endpoint de redimensionamento de imagens

Inicie a aplicação com python app.py e depois redimensione uma imagem 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

O resultado é sempre um PNG de 300 por 200. Esta operação estica a imagem; ela não preserva a proporção. Cada lado da saída deve ter entre 1 e 2048 pixels. Para miniaturas que preservam a proporção, escolha explicitamente uma política de ajuste ou recorte em vez de alterar silenciosamente a geometria solicitada.

Adicionando o endpoint de conversão de formato de imagem

/convert aceita PNG ou JPEG, sem diferenciar maiúsculas de minúsculas. O PNG mantém a transparência. O JPEG não consegue representar transparência, então o exemplo compõe os pixels transparentes sobre um fundo branco:

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

Testando a API com requisições de exemplo

Coloque estes testes em test_app.py e execute python -m unittest -v. O cliente de testes do Flask exercita o parser de requisições e o codificador do Pillow reais sem iniciar um 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()

Tratamento de erros e validação de entrada

Entradas inválidas esperadas recebem uma resposta 400 genérica; uma requisição HTTP grande demais recebe 413. Os limites de taxa retornam 429, e o limite de concorrência por processo retorna 503 sem enfileirar mais trabalho de imagem. Falhas inesperadas retornam uma resposta 500 sanitizada, nunca uma exceção do decodificador ou um stack trace.

Implantando a API em um servidor

Em um servidor Unix, um único worker do Gunicorn mantém os limites em memória desta demonstração em um só processo:

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

Uma implantação real também precisa de TLS em um proxy reverso, autenticação e autorização, prazos para as requisições, limites de corpo no proxy, monitoramento e workers isolados. CORS não é autenticação. Não confie em cabeçalhos de IP encaminhados até que a topologia do seu proxy impeça que os clientes os forneçam.

Antes de adicionar workers ou instâncias, configure um armazenamento compartilhado para o limite de taxa usando a documentação de armazenamento do Flask-Limiter. Contadores em memória são zerados ao reiniciar e não são compartilhados entre processos. Um timeout do servidor não é um mecanismo de cancelamento por operação para o Pillow.

Conclusão e próximos passos

Esta API mantém a validação de uploads e a codificação em um só lugar enquanto expõe duas operações. Amplie-a com testes para os formatos e a geometria de que você precisa e depois meça o uso de recursos com arquivos representativos. Para um fluxo de trabalho gerenciado, conheça o serviço de processamento de imagens da Transloadit em vez de operar sua própria frota de decodificadores.