Deploy a PyTorch API with FastAPI and Encore

PyTorch is an open-source machine learning framework used to build and train models for tasks such as image classification, language processing, and recommendation systems.

In this tutorial, we'll deploy a PyTorch image-classification model to your own AWS or GCP account using FastAPI and Encore, making it available through an API that applications can use to submit images and retrieve predictions.

Note: This tutorial uses Encore for Python, which is in private beta. Request access to follow along.

How the PyTorch API works

We'll use a pretrained MobileNetV3 model to classify uploaded images. Authenticated clients submit an image and receive a job ID. A Pub/Sub subscriber processes it in the background, and a status endpoint returns the predictions to the client that created the job.

Postgres stores the job before it is published. A recovery loop republishes pending work after a failed publish or interrupted worker. Each attempt claims the job with a token, which prevents an expired attempt from overwriting a later result.

Set up the project

Use Python 3.11 or later and a beta-enabled Encore application with a FastAPI service. Keep the encore.app, encore.service.json, and runtime configuration from your beta setup. Add these files alongside the service:

.
├── main.py
├── resources.py
├── security.py
├── jobs.py
├── inference.py
├── prepare_model.py
├── migrations/
│   └── 1_create_inference_jobs.up.sql
├── encore.app
├── encore.service.json
└── pyproject.toml

The examples use the encore-dev 2.0.0a8 Python API. Install FastAPI, PyTorch, Torchvision, Pillow, Uvicorn, and Pydantic. FastAPI file uploads also require python-multipart. The Encore distribution is named encore-dev and imported as encore.

Declare the backend resources

Create resources.py. The private image bucket holds uploads, and a separate versioned bucket holds model weights. The subscription passes only the job ID to the worker; the database supplies its storage key.

from encore import pubsub
from encore.secrets import Secret
from encore.storage import objects, postgres
from pydantic import BaseModel

MODEL_KEY = "mobilenet-v3-small-imagenet1k-v1.pt"

db = postgres.Database("inference", migrations="./migrations")
images = objects.Bucket("inference-images", public=False)
model_artifacts = objects.Bucket("model-artifacts", versioned=True, public=False)
api_keys = Secret[str]("InferenceAPIKeys")
inference_jobs = pubsub.Topic("inference-jobs")


class InferenceJob(BaseModel):
    job_id: str


async def classify_image(message: pubsub.Message[InferenceJob]) -> None:
    from inference import run_inference

    await run_inference(message.data.job_id)


inference_worker = pubsub.Subscription[InferenceJob](
    inference_jobs, "classify-images", handler=classify_image,
    max_retries=3, max_concurrency=2,
)

Encore uses these module-level declarations to provide local infrastructure and provision the cloud resources for deployment. Choose AWS or GCP in the environment settings; see infrastructure provisioning.

Create migrations/1_create_inference_jobs.up.sql:

CREATE TABLE inference_jobs (
    id UUID PRIMARY KEY,
    client_id TEXT NOT NULL,
    object_name TEXT NOT NULL,
    filename TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('uploading', 'queued', 'processing', 'completed', 'failed')
    ),
    attempts INTEGER NOT NULL DEFAULT 0,
    claim_token UUID,
    lease_until TIMESTAMPTZ,
    next_publish_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    predictions JSONB,
    error TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX inference_jobs_pending ON inference_jobs (next_publish_at)
    WHERE status IN ('queued', 'processing');

CREATE TABLE inference_budget (
    id INTEGER PRIMARY KEY CHECK (id = 1),
    minute_start TIMESTAMPTZ NOT NULL,
    minute_count INTEGER NOT NULL,
    day_start TIMESTAMPTZ NOT NULL,
    day_count INTEGER NOT NULL
);
INSERT INTO inference_budget VALUES
    (1, date_trunc('minute', now()), 0, date_trunc('day', now()), 0);

The budget row enforces a shared limit of ten submissions per minute and 100 per day. These are small tutorial limits; adjust them for the capacity and budget of your deployment. Every submitted job counts, including uploads that fail validation.

Authenticate clients and validate images

Configure the InferenceAPIKeys secret with a JSON object mapping client IDs to different, randomly generated API keys. Use at least 32 random bytes per key and keep them in server-side clients. Set the secret separately for local development and production through Encore's secret management.

Create security.py:

import asyncio
import json
import secrets
import warnings
from contextlib import contextmanager
from io import BytesIO
from typing import Annotated

from fastapi import Depends, HTTPException
from fastapi.security import APIKeyHeader
from PIL import Image, UnidentifiedImageError

from resources import api_keys

MAX_UPLOAD_BYTES = 10 * 1024 * 1024
MAX_IMAGE_PIXELS = 4_000_000
image_slots = asyncio.BoundedSemaphore(1)
image_tasks: set[asyncio.Task] = set()
key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


def client_keys() -> dict[str, str]:
    keys = json.loads(api_keys.value)
    if not isinstance(keys, dict) or not keys or not all(
        isinstance(client, str) and isinstance(key, str) and len(key) >= 32
        for client, key in keys.items()
    ):
        raise RuntimeError("Configure InferenceAPIKeys with client IDs and strong keys")
    if len(set(keys.values())) != len(keys):
        raise RuntimeError("Give each client a different API key")
    return keys


def require_client(
    key: Annotated[str | None, Depends(key_header)],
) -> str:
    supplied = (key or "").encode("utf-8")
    for client_id, expected in client_keys().items():
        if secrets.compare_digest(supplied, expected.encode("utf-8")):
            return client_id
    raise HTTPException(401, "Invalid API key")


async def take_image_slot_nowait() -> None:
    if image_slots.locked():
        raise HTTPException(503, "Image processing is busy; retry later")
    # No suspension: Semaphore.acquire takes an available permit immediately.
    # release() reserves permits for queued waiters before waking them.
    await image_slots.acquire()


def release_image_slot() -> None:
    image_slots.release()


async def run_image_task(fn, *args, wait: bool = True, slot_acquired: bool = False):
    if not slot_acquired:
        if wait:
            await image_slots.acquire()
        else:
            await take_image_slot_nowait()
    try:
        task = asyncio.create_task(asyncio.to_thread(fn, *args))
    except BaseException:
        release_image_slot()
        raise
    image_tasks.add(task)

    def finished(done):
        image_tasks.discard(done)
        release_image_slot()
        if not done.cancelled():
            done.exception()

    task.add_done_callback(finished)
    return await asyncio.shield(task)


@contextmanager
def open_image(data: bytes):
    try:
        with warnings.catch_warnings():
            warnings.simplefilter("error", Image.DecompressionBombWarning)
            with Image.open(BytesIO(data)) as image:
                if image.width * image.height > MAX_IMAGE_PIXELS:
                    raise HTTPException(413, "Image dimensions are too large")
                if image.format not in {"JPEG", "PNG"} or getattr(image, "n_frames", 1) != 1:
                    raise HTTPException(415, "Upload a single JPEG or PNG image")
                image.verify()
            with Image.open(BytesIO(data)) as image:
                yield image
    except (Image.DecompressionBombError, Image.DecompressionBombWarning):
        raise HTTPException(413, "Image dimensions are too large") from None
    except (UnidentifiedImageError, OSError, SyntaxError, ValueError):
        raise HTTPException(415, "Invalid image") from None


def validate_image(data: bytes) -> str:
    with open_image(data) as image:
        image.load()
        return Image.MIME[image.format]

Clients send their key in X-API-Key over HTTPS. The client ID is stored with each job and checked when its result is requested. Uploads must be single-frame JPEG or PNG images with no more than four million pixels, such as a 2000-by-2000 image. Pillow inspects the bytes to determine the content type.

Each process runs one image-validation or inference task at a time. The upload endpoint acquires the slot without waiting, before reserving a job or reading the file. Busy uploads receive 503; subscriber tasks wait for a slot. Releasing the semaphore reserves its permit for the next waiting task, so a new upload cannot take it during handoff. Cancelling a request leaves the slot reserved until its processing thread finishes.

Reserve jobs and recover interrupted work

Create jobs.py. Reserving a job updates the shared quota and inserts the job in one SQL statement. A rejected request cannot reserve a job, and concurrent application instances use the same budget row.

After the image is stored, the API marks the job queued and tries to publish it. If publishing fails, the queued row remains available to recovery. The recovery loop also finds expired worker claims, stops exhausted jobs, and removes terminal jobs and their images after a day.

import asyncio
import logging
from contextlib import suppress

from encore.storage import objects
from fastapi import HTTPException

from resources import InferenceJob, db, images, inference_jobs

log = logging.getLogger(__name__)


async def reserve_job(job_id: str, client_id: str, filename: str) -> str:
    object_name = f"inputs/{job_id}"
    row = await db.query(
        """
        WITH admitted AS (
            UPDATE inference_budget SET
                minute_start = date_trunc('minute', now()),
                minute_count = CASE
                    WHEN minute_start = date_trunc('minute', now())
                    THEN minute_count + 1 ELSE 1 END,
                day_start = date_trunc('day', now()),
                day_count = CASE WHEN day_start = date_trunc('day', now())
                    THEN day_count + 1 ELSE 1 END
            WHERE id = 1
              AND (minute_start < date_trunc('minute', now()) OR minute_count < 10)
              AND (day_start < date_trunc('day', now()) OR day_count < 100)
            RETURNING id
        )
        INSERT INTO inference_jobs (id, client_id, object_name, filename, status)
        SELECT $1::uuid, $2, $3, $4, 'uploading' FROM admitted
        RETURNING object_name
        """,
        job_id, client_id, object_name, filename[:255],
    ).one()
    if row is None:
        raise HTTPException(429, "Upload quota reached; try again later")
    return row["object_name"]


async def publish_job(job_id: str) -> None:
    try:
        await inference_jobs.publish(InferenceJob(job_id=job_id))
    except Exception:
        log.exception("Publish failed; recovery will retry job %s", job_id)


async def recover_once() -> None:
    await db.exec(
        """
        UPDATE inference_jobs SET status = 'failed', claim_token = NULL,
            error = 'Job expired or exhausted its attempts', updated_at = now()
        WHERE (status = 'queued' OR (status = 'processing' AND lease_until < now()))
          AND (attempts >= 3 OR created_at < now() - interval '1 hour')
        """
    )
    await db.exec(
        """
        UPDATE inference_jobs SET status = 'failed', error = 'Upload interrupted',
            updated_at = now()
        WHERE status = 'uploading' AND created_at < now() - interval '10 minutes'
        """
    )
    pending = await db.query(
        """
        WITH due AS (
            SELECT id FROM inference_jobs
            WHERE (status = 'queued' OR
                   (status = 'processing' AND lease_until < now()))
              AND attempts < 3 AND next_publish_at <= now()
              AND created_at >= now() - interval '1 hour'
            ORDER BY next_publish_at LIMIT 20 FOR UPDATE SKIP LOCKED
        )
        UPDATE inference_jobs AS j
        SET next_publish_at = now() + interval '30 seconds'
        FROM due WHERE j.id = due.id RETURNING j.id::text AS id
        """
    ).all()
    for row in pending:
        await publish_job(row["id"])

    expired = await db.query(
        """
        SELECT id::text, object_name FROM inference_jobs
        WHERE status IN ('completed', 'failed')
          AND updated_at < now() - interval '1 day'
        LIMIT 50
        """
    ).all()
    for row in expired:
        with suppress(objects.ObjectNotFound):
            await images.remove(row["object_name"])
        await db.exec("DELETE FROM inference_jobs WHERE id = $1::uuid", row["id"])


async def recovery_loop() -> None:
    while True:
        try:
            await recover_once()
        except Exception:
            log.exception("Job recovery failed; retrying")
        await asyncio.sleep(15)

The recovery loop runs inside the application, including when no requests arrive. Keep at least one instance running with CPU available for background work. For Cloud Run, this means instance-based billing and a minimum instance count of at least one.

Load the model and process an image

Create inference.py. The model loads once per process from the weights bucket. PyTorch uses CPU inference. The worker checks the stored image and decodes it once during RGB conversion; it does not repeat the upload handler's full validation decode.

Start with one application process and 2 GiB of memory, then check peak memory usage with your images before increasing concurrency. Each additional process loads its own model and has its own image-processing slot.

import asyncio
import json
import logging
from io import BytesIO
from uuid import uuid4

import torch
from torchvision.models import MobileNet_V3_Small_Weights, mobilenet_v3_small

from resources import MODEL_KEY, db, images, model_artifacts
from security import open_image, run_image_task

log = logging.getLogger(__name__)
model_lock = asyncio.Lock()
loaded_model = None


def load_model(data: bytes):
    model = mobilenet_v3_small(weights=None)
    model.load_state_dict(torch.load(BytesIO(data), map_location="cpu", weights_only=True))
    return model.eval(), MobileNet_V3_Small_Weights.IMAGENET1K_V1.meta["categories"]


async def get_model():
    global loaded_model
    async with model_lock:
        if loaded_model is None:
            data = await model_artifacts.download(MODEL_KEY)
            loaded_model = await asyncio.to_thread(load_model, data)
    return loaded_model


def predict(model, categories, image_data: bytes) -> list[dict]:
    weights = MobileNet_V3_Small_Weights.IMAGENET1K_V1
    with open_image(image_data) as image:
        tensor = weights.transforms()(image.convert("RGB")).unsqueeze(0)
    with torch.inference_mode():
        probabilities = model(tensor).softmax(dim=1)[0]
    scores, indices = probabilities.topk(5)
    return [
        {"label": categories[index], "confidence": round(float(score), 4)}
        for score, index in zip(scores.tolist(), indices.tolist(), strict=True)
    ]


async def run_inference(job_id: str) -> None:
    token = str(uuid4())
    job = await db.query(
        """
        UPDATE inference_jobs SET status = 'processing', attempts = attempts + 1,
            claim_token = $2::uuid, lease_until = now() + interval '5 minutes',
            updated_at = now()
        WHERE id = $1::uuid AND attempts < 3
          AND created_at >= now() - interval '1 hour'
          AND (status = 'queued' OR (status = 'processing' AND lease_until < now()))
        RETURNING object_name, attempts
        """, job_id, token,
    ).one()
    if job is None:
        return
    try:
        image_data = await images.download(job["object_name"])
        model, categories = await get_model()
        predictions = await run_image_task(predict, model, categories, image_data)
    except Exception:
        log.exception("Inference attempt failed for job %s", job_id)
        await db.exec(
            """
            UPDATE inference_jobs SET status = CASE WHEN attempts >= 3
                THEN 'failed' ELSE 'queued' END,
                error = CASE WHEN attempts >= 3 THEN 'Inference failed' ELSE NULL END,
                claim_token = NULL, lease_until = NULL, updated_at = now(),
                next_publish_at = now() + interval '30 seconds'
            WHERE id = $1::uuid AND claim_token = $2::uuid AND status = 'processing'
            """, job_id, token,
        )
        return
    await db.exec(
        """
        UPDATE inference_jobs SET status = 'completed', predictions = $3::jsonb,
            error = NULL, claim_token = NULL, lease_until = NULL, updated_at = now()
        WHERE id = $1::uuid AND claim_token = $2::uuid AND status = 'processing'
        """, job_id, token, json.dumps(predictions),
    )

The claim update is atomic: only a queued job or an expired claim can start an attempt. A duplicate message arriving during a valid claim returns without running inference again. Completed and failed jobs cannot be claimed.

Attempts are counted in Postgres, including an attempt interrupted by a process crash. Recoverable failures return to queued; the third failed attempt becomes failed. A worker that disappears leaves a five-minute lease, after which recovery can publish the job again. Choose a lease longer than expected inference time. The token checks prevent an older worker from overwriting a newer attempt's result.

Add the FastAPI endpoints

Create main.py. Authentication runs before a job is reserved. The upload endpoint reads at most 10 MB plus one byte, validates the image, and stores it under a generated key. The status endpoint validates the path as a UUID and restricts the lookup to the authenticated client.

import asyncio
from contextlib import asynccontextmanager, suppress
from typing import Annotated, Literal
from uuid import UUID, uuid4

import encore
from fastapi import Depends, FastAPI, File, HTTPException, UploadFile
from pydantic import BaseModel

from jobs import publish_job, recovery_loop, reserve_job
from resources import db, images
from security import (MAX_UPLOAD_BYTES, client_keys, require_client,
                      run_image_task, validate_image, take_image_slot_nowait,
                      release_image_slot)


@asynccontextmanager
async def lifespan(_: FastAPI):
    client_keys()
    await encore.start()
    recovery = asyncio.create_task(recovery_loop())
    try:
        yield
    finally:
        recovery.cancel()
        with suppress(asyncio.CancelledError):
            await recovery


app = FastAPI(lifespan=lifespan)
Client = Annotated[str, Depends(require_client)]


class JobResult(BaseModel):
    id: UUID
    filename: str
    status: Literal['uploading', 'queued', 'processing', 'completed', 'failed']
    predictions: list[dict] | None = None
    error: str | None = None


@app.post("/inferences", status_code=202)
async def create_inference(client_id: Client, file: UploadFile = File(...)) -> dict:
    job_id = str(uuid4())
    slot_owned = False
    job_reserved = False
    try:
        await take_image_slot_nowait()
        slot_owned = True
        object_name = await reserve_job(job_id, client_id, file.filename or "upload")
        job_reserved = True
        image_data = await file.read(MAX_UPLOAD_BYTES + 1)
        if not image_data:
            raise HTTPException(400, "The uploaded file is empty")
        if len(image_data) > MAX_UPLOAD_BYTES:
            raise HTTPException(413, "Images must be 10 MB or smaller")
        slot_owned = False  # The processing task now owns and releases the slot.
        content_type = await run_image_task(validate_image, image_data, slot_acquired=True)
        await images.upload(object_name, image_data, content_type=content_type,
                            if_not_exists=True)
        queued = await db.query(
            "UPDATE inference_jobs SET status = 'queued', updated_at = now() "
            "WHERE id = $1::uuid AND status = 'uploading' RETURNING id", job_id,
        ).one()
        if queued is None:
            raise HTTPException(409, "Upload reservation expired; submit a new job")
    except Exception:
        if job_reserved:
            await db.exec(
                "UPDATE inference_jobs SET status = 'failed', error = 'Upload failed', "
                "updated_at = now() WHERE id = $1::uuid AND status = 'uploading'", job_id,
            )
        raise
    finally:
        if slot_owned:
            release_image_slot()
        await file.close()
    await publish_job(job_id)
    return {"id": job_id, "status": "queued"}


@app.get("/inferences/{job_id}", response_model=JobResult)
async def get_inference(job_id: UUID, client_id: Client) -> JobResult:
    row = await db.query(
        """
        SELECT id::text, filename, status, predictions, error FROM inference_jobs
        WHERE id = $1::uuid AND client_id = $2
        """, str(job_id), client_id,
    ).one()
    if row is None:
        raise HTTPException(404, "Inference job not found")
    return JobResult(**row)

The lifespan hook starts Encore's subscription handler and the recovery loop. The loop retries after temporary errors and is cancelled during application shutdown. A process stopped during inference is recovered through the persisted job claim.

For an internet-facing deployment, also set a request-body limit at the gateway or reverse proxy. Allow a little room above the image limit for multipart headers; 12 MB is sufficient for this example. That bounds the multipart request before FastAPI parses it.

Prepare the model weights

Create prepare_model.py. It downloads the named Torchvision weights, writes the state dictionary to a file, and uploads it to the local model bucket:

import asyncio
from pathlib import Path

import torch
from torchvision.models import MobileNet_V3_Small_Weights, mobilenet_v3_small

from resources import MODEL_KEY, model_artifacts


def prepare_weights() -> bytes:
    model = mobilenet_v3_small(weights=MobileNet_V3_Small_Weights.IMAGENET1K_V1)
    path = Path(MODEL_KEY)
    torch.save(model.state_dict(), path)
    return path.read_bytes()


async def main():
    data = await asyncio.to_thread(prepare_weights)
    if not await model_artifacts.exists(MODEL_KEY):
        await model_artifacts.upload(
            MODEL_KEY, data, content_type="application/octet-stream", if_not_exists=True,
        )


if __name__ == "__main__":
    asyncio.run(main())

Run it with the application's local infrastructure configured:

encore exec -- python prepare_model.py

The script-running guide explains this command. Before accepting production jobs, upload the same generated .pt file under mobilenet-v3-small-imagenet1k-v1.pt in that environment's model-artifacts bucket. Local bucket contents are separate from production. Only trusted administrators should be able to replace model weights.

Run locally

After setting the local API-key secret and preparing the weights, start the application:

encore run

Set INFERENCE_API_KEY in your client shell to one of the configured keys, then submit an image:

curl -X POST http://127.0.0.1:4000/inferences \
  -H "X-API-Key: $INFERENCE_API_KEY" \
  -F "file=@cat.jpg"

The response is 202 Accepted with a job ID. Request its status using the same key:

curl http://127.0.0.1:4000/inferences/JOB_ID \
  -H "X-API-Key: $INFERENCE_API_KEY"

A completed job includes its predictions:

{
  "id": "57bd03f4-ecdf-47f2-a9e7-28a7fd83baad",
  "filename": "cat.jpg",
  "status": "completed",
  "predictions": [
    {"label": "tabby", "confidence": 0.6241},
    {"label": "tiger cat", "confidence": 0.1948}
  ],
  "error": null
}

Deploy to AWS or GCP

1. Connect your AWS or GCP account to the application in the Encore dashboard.
2. Create a production environment, configure its API-key secret, and select the region, CPU, memory, and always-running capacity in the infrastructure settings. Prepare the production model bucket before enabling uploads.
3. Choose the deployment branch, then push the code. The deployment guide covers repository setup and deployment options.

Encore builds the application, provisions its database, buckets, and Pub/Sub resources in your cloud account, and deploys the service. This example uses CPU compute; a GPU model requires a deployment with suitable GPU support.

Encore for Python is in private beta. Request access for help setting up and deploying your Python application.

Related reading

• Encore for Python
• Message queues vs. Pub/Sub
• What is Pub/Sub?
• Understanding Encore