Creating a simple image processing API with Python and Flask

Kevin van Zonneveld

Co-founder · Amsterdam, The Netherlands ·

A small image API helps you learn how uploads, decoders and HTTP responses fit together. Here we build two endpoints with Flask and Pillow: one stretches an image to a requested size, and the other converts it to PNG or JPEG. Both use the same validation and encoding path.

This is a loopback development service, not an authenticated public upload service. File-size, pixel and concurrency limits reduce resource use; they do not make a native decoder a sandbox.

Introduction to image processing APIs

The request contains one multipart file plus operation parameters. The server checks the decoded format and dimensions, performs the operation and returns image bytes. Filenames and browser-supplied MIME types are not evidence that an upload is safe.

Setting up the development environment

Use Python 3.12 or newer. In a new directory, create requirements.txt:

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

Then install into a virtual environment:

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

On Windows, activate the environment with .venv\Scripts\activate. Keep resolved dependency versions in your deployment lockfile and review updates before exposing a service to uploads.

Creating a basic Flask application

Put this complete application in app.py. There is one startup block, after all routes and handlers:

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)

Integrating pillow for image processing

load_image() opens only JPEG and PNG decoders. It rejects animated PNGs and checks decoded pixel count before loading the image. It applies EXIF orientation and creates RGBA pixels for the shared processing path. The output deliberately drops source metadata, including embedded GPS data.

The 8 MiB file limit is separate from the multipart request limit. Neither compressed file size nor dimensions alone bound every decoder's CPU and memory use. Keep Pillow current and isolate public processing workers with operating-system resource limits.

Implementing image resizing endpoint

Start the application with python app.py, then resize a local image:

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

The result is always a 300 by 200 PNG. This operation stretches the image; it does not preserve aspect ratio. Each output side must be between 1 and 2048 pixels. For aspect-preserving thumbnails, choose a fit or crop policy explicitly instead of silently changing the requested geometry.

Adding image format conversion endpoint

/convert accepts PNG or JPEG, case-insensitively. PNG retains transparency. JPEG cannot represent transparency, so the example composites transparent pixels onto white:

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

Testing the API with sample requests

Put these tests in test_app.py and run python -m unittest -v. Flask's test client exercises the actual request parser and Pillow encoder without starting a server:

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

Error handling and input validation

Expected invalid inputs receive a generic 400 response; an oversized HTTP request receives 413. Rate limits return 429, and the per-process concurrency cap returns 503 without queuing more image work. Unexpected failures return a sanitized 500 response, never a decoder exception or stack trace.

Deploying the API to a server

On a Unix server, a single Gunicorn worker keeps this demo's in-memory limits in one process:

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

A real deployment also needs TLS at a reverse proxy, authentication and authorization, request deadlines, body limits at the proxy, monitoring and isolated workers. CORS is not authentication. Do not trust forwarded IP headers until your proxy topology prevents clients from supplying them.

Before adding workers or instances, configure a shared rate-limit store using the Flask-Limiter storage documentation. Memory-backed counters reset on restart and are not shared across processes. A server timeout is not a per-operation cancellation mechanism for Pillow.

Conclusion and next steps

This API keeps upload validation and encoding in one place while exposing two operations. Extend it with tests for your required formats and geometry, then measure resource use with representative files. For a managed workflow, explore Transloadit's image processing service instead of running your own decoder fleet.