Criar API simples de processamento de 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 integram. Aqui, criamos dois endpoints com Flask e Pillow: um estica a imagem até o tamanho solicitado, e o outro a converte para PNG ou JPEG. Ambos usam o mesmo fluxo de validação e codificação.

Este é um serviço de desenvolvimento em loopback, não um serviço público de upload com autenticação. Os limites de tamanho de arquivo, pixels e concorrência reduzem o uso de recursos; eles não transformam um decodificador nativo em um ambiente isolado.

Introdução às APIs de processamento de imagens

A requisição contém um arquivo multipart e 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 arquivos e tipos MIME fornecidos pelo navegador não são evidência de que um upload é seguro.

Configurar o ambiente de desenvolvimento

Use Python 3.12 ou mais recente, Bash e cURL no Linux. Verifique python3 --version e confirme que curl --help all lista --fail-with-body antes de criar arquivos. 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

Em seguida, instale em um ambiente virtual:

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

Se a criação do ambiente ou a instalação falhar, corrija o erro e repita o bloco de configuração no mesmo diretório; mantenha requirements.txt e seus arquivos de código-fonte. Os comandos param na etapa que falhou. O ambiente virtual mantém os pacotes deste exemplo separados dos pacotes de um projeto Python que o contenha.

Criar uma aplicação Flask básica

Coloque esta aplicação completa em app.py. Há um único bloco de inicialização, após todas as rotas e funções de tratamento:

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 o Pillow para processamento de imagens

load_image() abre apenas decodificadores JPEG e PNG. Este exemplo aceita PNGs com até oito bits por canal e rejeita PNGs de 16 bits em vez de alterar silenciosamente seu intervalo numérico. Ele rejeita PNGs animados e verifica a contagem de pixels decodificados antes de carregar a imagem. Aplica a orientação EXIF e cria pixels RGBA para o fluxo de processamento compartilhado. A saída elimina deliberadamente os metadados de origem, incluindo dados GPS incorporados. Ela não aplica perfis de cor ICC; primeiro, converta as entradas com gerenciamento de cores para sRGB.

A documentação de modos de imagem do Pillow explica a representação RGBA de oito bits e seus limites.

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

Implementar o endpoint de redimensionamento de imagens

Inicie a aplicação com python app.py. Deixe-a em execução e abra outro terminal no mesmo diretório para usar o cURL. Coloque ali um JPEG local chamado photo.jpg e redimensione-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

Em caso de sucesso, este comando retorna 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 preservem a proporção, escolha explicitamente uma política de ajuste ou recorte, em vez de alterar silenciosamente a geometria solicitada.

Abra resized.png para verificar seu conteúdo. O cURL substitui o arquivo de saída indicado quando é executado novamente com sucesso; com --fail-with-body, um erro HTTP pode gravar JSON nele, então verifique também o status de término do cURL.

Adicionar o endpoint de conversão de formato de imagem

/convert aceita PNG ou JPEG, sem distinguir maiúsculas de minúsculas. PNG mantém a transparência. JPEG não pode representar transparência, então o exemplo compõe os pixels transparentes sobre um fundo branco. Coloque um PNG de oito bits chamado transparent.png no mesmo diretório e converta-o:

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

Abra converted.jpg: as áreas transparentes devem estar brancas, e os pixels parcialmente transparentes devem se mesclar com o branco. Ambos os endpoints retornam imagens diretamente; a aplicação não salva uploads nem resultados no servidor. Pare o servidor com Ctrl+C quando terminar.

Testar 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 analisador de requisições e o codificador 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"
        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()

Tratamento de erros e validação de entrada

Entradas inválidas previstas recebem uma resposta 400 genérica; uma requisição HTTP acima do limite de tamanho recebe 413. Cada endpoint permite 60 requisições por minuto por IP de cliente. Os limites de taxa retornam 429, e uma requisição que chega à aplicação enquanto as duas vagas de processamento estão ocupadas retorna 503. Um servidor HTTP ainda pode enfileirar conexões antes que elas cheguem à aplicação. Falhas inesperadas retornam uma resposta 500 sem detalhes sensíveis, nunca uma exceção do decodificador ou um rastreamento de stack. Um arquivo acima de 8 MiB dentro do limite da requisição HTTP recebe 400; os limites de uso de recursos do Flask podem produzir 413 antes disso para a requisição, os campos do formulário ou a contagem de partes.

Executar a API local com Gunicorn

No Linux, um único processo de trabalho 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 \
  --no-control-socket app:app

Mantenha este serviço escutando em loopback. Configure separadamente um ponto de entrada HTTPS com autenticação e o isolamento dos processos de trabalho antes de aceitar uploads remotos. O socket de gerenciamento opcional está desativado para que esta demonstração não compartilhe o caminho de socket padrão do Gunicorn com outra instância local. Ctrl+C para o servidor; o encerramento dos processos de trabalho pode levar até o período de tolerância padrão de 30 segundos.

Antes de adicionar processos de trabalho ou instâncias, configure um armazenamento compartilhado para os limites de taxa usando a documentação de armazenamento do Flask-Limiter. Os contadores mantidos em memória são zerados na reinicialização e não são compartilhados entre processos. Um tempo limite do servidor não é um mecanismo de cancelamento por operação do Pillow.

Conclusão e próximos passos

Amplie os testes para os formatos e a geometria de que você precisa e, em seguida, meça o uso de recursos com arquivos representativos. Para um fluxo de trabalho gerenciado, explore o serviço de processamento de imagens da Transloadit em vez de operar sua própria infraestrutura de decodificadores.