How to Integrate Filestack with FastAPI Using the Python SDK

Posted on

FastAPI is async-first, which means dropping a synchronous SDK into an async def route can collapse your concurrency if you’re not careful. The filestack-python SDK — the Python client for Filestack’s file uploader API — is synchronous, so the integration pattern with FastAPI is slightly different from Django or Flask: you wrap blocking SDK calls in run_in_threadpool and treat them like any other I/O-bound dependency.

There is no fastapi-filestack package on PyPI, but FastAPI’s UploadFile is built on a SpooledTemporaryFile that the Filestack SDK accepts directly through its .file attribute. That means the integration is mostly a service module, a couple of dependency-injected helpers, and a webhook route that knows how to read raw bytes.

Every snippet here runs against the real SDK. Copy them in order and you’ll have working async uploads, signed downloads, transformations, dependency injection, and verified webhooks by the end.

Key takeaways

  • Filestack has no FastAPI-specific package, but filestack-python works inside any FastAPI route through run_in_threadpool
  • UploadFile.file is a SpooledTemporaryFile that you can pass straight to client.upload(file_obj=…) without staging to disk
  • Wrapping the sync SDK in run_in_threadpool keeps the event loop responsive during uploads
  • Dependency injection is the natural place to manage Client and Security lifecycles in FastAPI
  • Webhook routes must read await request.body() as raw bytes before parsing to keep signature verification working

Before you start

You need:

Python 3.8 or higher

FastAPI 0.100+ and an ASGI server like Uvicorn

A Filestack account for your API key and app secret

Familiarity with async/await, dependency injection, and Pydantic settings

Pull your API key and app secret from the Filestack developer portal. The API key is fine in your frontend. The app secret is server-only and signs every security policy.

Step 1: Install and configure

Install the SDK alongside FastAPI:

pip install filestack-python fastapi uvicorn pydantic-settings python-multipart

python-multipart is required for FastAPI to parse UploadFile form data. pydantic-settings handles config loading from environment variables.

Build a settings class:

# config.py

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):

    filestack_api_key: str

    filestack_app_secret: str

    filestack_webhook_secret: str = ""

    model_config = SettingsConfigDict(env_file=".env", env_prefix="")


settings = Settings()

And a .env file:

FILESTACK_API_KEY=Axxxxxxxxxxxxxxxxxxxxx

FILESTACK_APP_SECRET=your-app-secret-here

FILESTACK_WEBHOOK_SECRET=your-webhook-secret

Step 2: Build the service layer

Putting SDK calls directly in route handlers makes them harder to test and ties your business logic to Filestack. A service module isolates the integration and gives you one place to manage Client and Security objects.

Create services/filestack_service.py:

# services/filestack_service.py

import time

from typing import Optional

from filestack import Client, Filelink, Security

from fastapi import UploadFile

from config import settings


def build_security(

    expiry_seconds: int = 3600,

    calls: Optional[list[str]] = None,

    handle: Optional[str] = None,

) -> Security:

    """Build a Security object scoped to specific calls and (optionally) a handle."""

    policy = {

        "expiry": int(time.time()) + expiry_seconds,

        "call": calls or ["pick", "read", "store", "remove", "convert"],

    }

    if handle:

        policy["handle"] = handle

    return Security(policy, settings.filestack_app_secret)


def build_client(with_security: bool = True) -> Client:

    """Return a Filestack Client, optionally with a signing policy attached."""

    if with_security:

        return Client(settings.filestack_api_key, security=build_security())

    return Client(settings.filestack_api_key)


def _upload_sync(file_obj, filename: str, content_type: str, path: str) -> Filelink:

    """Synchronous upload, called via run_in_threadpool from async routes."""

    client = build_client(with_security=True)

    store_params = {

        "filename": filename,

        "mimetype": content_type,

        "path": path,

    }

    return client.upload(file_obj=file_obj, store_params=store_params)


def _signed_url_sync(handle: str, expiry_seconds: int = 3600) -> str:

    """Synchronous signed URL builder, called via run_in_threadpool."""

    security = build_security(expiry_seconds=expiry_seconds, calls=["read", "convert"], handle=handle)

    return Filelink(handle, security=security).signed_url()


def _delete_sync(handle: str) -> None:

    """Synchronous delete, called via run_in_threadpool."""

    security = build_security(calls=["remove"], handle=handle)

    Filelink(handle, apikey=settings.filestack_api_key, security=security).delete()


def _metadata_sync(handle: str, fields: Optional[list[str]] = None) -> dict:

    """Synchronous metadata fetch, called via run_in_threadpool."""

    filelink = Filelink(handle, apikey=settings.filestack_api_key)

    return filelink.metadata(fields or ["size", "mimetype", "filename"])


def _thumbnail_url_sync(handle: str, width: int = 200, height: int = 200) -> str:

    """Synchronous transformation URL builder."""

    security = build_security(calls=["read", "convert"], handle=handle)

    filelink = Filelink(handle, apikey=settings.filestack_api_key, security=security)

    return filelink.resize(width=width, height=height, fit="crop").enhance().url

The _sync suffix on the inner functions is intentional. These are the blocking entry points that get scheduled onto the threadpool. The async wrappers come next.

Step 3: Add async wrappers with run_in_threadpool

filestack-python makes blocking HTTP calls. Calling it directly inside an async def route would freeze the event loop and tank your concurrency. FastAPI ships with run_in_threadpool exactly for this case.

Add to services/filestack_service.py:

# services/filestack_service.py (continued)

from fastapi.concurrency import run_in_threadpool


async def upload_uploadfile(file: UploadFile, owner_id: int) -> Filelink:

    """Upload a FastAPI UploadFile to Filestack without blocking the event loop."""

    # Seek to the start in case the file was already read for validation

    await file.seek(0)

    return await run_in_threadpool(

        _upload_sync,

        file.file,  # the underlying SpooledTemporaryFile

        file.filename or "unnamed",

        file.content_type or "application/octet-stream",

        f"user-uploads/{owner_id}/",

    )


async def signed_url_for(handle: str, expiry_seconds: int = 3600) -> str:

    return await run_in_threadpool(_signed_url_sync, handle, expiry_seconds)


async def delete_handle(handle: str) -> None:

    return await run_in_threadpool(_delete_sync, handle)


async def fetch_metadata(handle: str, fields: Optional[list[str]] = None) -> dict:

    return await run_in_threadpool(_metadata_sync, handle, fields)


async def thumbnail_url(handle: str, width: int = 200, height: int = 200) -> str:

    return await run_in_threadpool(_thumbnail_url_sync, handle, width, height)

Two details worth noting. First, UploadFile.file is the actual SpooledTemporaryFile underneath. The SDK accepts it directly because it exposes a standard file-like interface. Second, await file.seek(0) is defensive. If any middleware or earlier dependency read the file (for example, to check MIME type), the cursor would be at the end and the upload would send zero bytes.

Step 4: Wire up an upload route

Build the upload endpoint. This is where the async wrappers pay off.

# routers/uploads.py

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status

from sqlalchemy.orm import Session

from db import get_db

from auth import get_current_user, User

from models import Document

from services import filestack_service

router = APIRouter(prefix="/api/uploads", tags=["uploads"])


@router.post("/", status_code=status.HTTP_201_CREATED)

async def create_upload(

    file: UploadFile = File(...),

    current_user: User = Depends(get_current_user),

    db: Session = Depends(get_db),

):

    if not file.filename:

        raise HTTPException(status_code=400, detail="Empty filename")

    filelink = await filestack_service.upload_uploadfile(file, owner_id=current_user.id)

    # Reading size from the spooled file (after upload, the cursor is at the end)

    file.file.seek(0, 2)  # SEEK_END on the underlying SpooledTemporaryFile

    size = file.file.tell()

    doc = Document(

        owner_id=current_user.id,

        filestack_handle=filelink.handle,

        filestack_url=filelink.url,

        original_name=file.filename,

        size_bytes=size,

    )

    db.add(doc)

    db.commit()

    db.refresh(doc)

    return {

        "id": doc.id,

        "handle": filelink.handle,

        "url": filelink.url,

    }

A SQLAlchemy model to back it (your ORM choice is irrelevant to Filestack):

# models.py

from datetime import datetime

from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, Integer, String

from db import Base


class Document(Base):

    __tablename__ = "documents"

    id = Column(Integer, primary_key=True, index=True)

    owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    filestack_handle = Column(String(64), unique=True, nullable=False, index=True)

    filestack_url = Column(String(500), nullable=False)

    original_name = Column(String(255), nullable=False)

    size_bytes = Column(BigInteger, default=0)

    created_at = Column(DateTime, default=datetime.utcnow)

Store the handle. It’s the durable identifier you need for fresh signed URLs, transformations, and deletions. The URL is convenience metadata.

Step 5: Serve files with signed URLs

Public files can use filelink.url directly. For anything private, generate a short-lived signed URL on demand.

# routers/uploads.py (continued)

from fastapi import HTTPException


@router.get("/{doc_id}/download-url")

async def get_download_url(

    doc_id: int,

    current_user: User = Depends(get_current_user),

    db: Session = Depends(get_db),

):

    doc = db.query(Document).filter(Document.id == doc_id).first()

    if not doc:

        raise HTTPException(status_code=404, detail="Document not found")

    if doc.owner_id != current_user.id:

        raise HTTPException(status_code=403, detail="Forbidden")

    url = await filestack_service.signed_url_for(doc.filestack_handle, expiry_seconds=300)

    return {"url": url}

The frontend hits this endpoint, gets a five-minute URL, and either redirects or fetches the file. After the expiry, the URL stops working. Leaked links go stale fast.

Step 6: Use dependency injection for cleaner reuse

If you want a more idiomatic FastAPI shape, expose Filestack as a dependency. This makes mocking trivial in tests and gives you a single place to manage client lifecycle.

# dependencies.py

from typing import AsyncGenerator

from filestack import Client

from services.filestack_service import build_client


async def get_filestack_client() -> AsyncGenerator[Client, None]:

    """Yields a Filestack Client per request. Override in tests with app.dependency_overrides."""

    client = build_client(with_security=True)

    try:

        yield client

    finally:

        # Filestack Client has no explicit close, but this is where you'd add it if it did

        pass

Use it in a route:

# routers/uploads.py

from fastapi.concurrency import run_in_threadpool

from filestack import Client

from dependencies import get_filestack_client


@router.post("/v2", status_code=201)

async def create_upload_v2(

    file: UploadFile = File(...),

    current_user: User = Depends(get_current_user),

    fs: Client = Depends(get_filestack_client),

):

    await file.seek(0)

    def _do_upload():

        return fs.upload(

            file_obj=file.file,

            store_params={

                "filename": file.filename,

                "mimetype": file.content_type,

                "path": f"user-uploads/{current_user.id}/",

            },

        )

    filelink = await run_in_threadpool(_do_upload)

    return {"handle": filelink.handle, "url": filelink.url}

In tests:

# tests/test_uploads.py

from unittest.mock import MagicMock

from app import app

from dependencies import get_filestack_client


def override_filestack_client():

    fake = MagicMock()

    fake.upload.return_value.handle = "test-handle"

    fake.upload.return_value.url = "https://cdn.filestackcontent.com/test-handle"

    yield fake

app.dependency_overrides[get_filestack_client] = override_filestack_client

No network calls in tests, no real handles burned, no flakiness.

Step 7: Apply transformations on the fly

Filestack transformations run at delivery time. Chain them on a Filelink and return the URL.

# routers/uploads.py (continued)

@router.get("/{doc_id}/thumbnail")

async def get_thumbnail(

    doc_id: int,

    width: int = 200,

    height: int = 200,

    current_user: User = Depends(get_current_user),

    db: Session = Depends(get_db),

):

    doc = db.query(Document).filter(Document.id == doc_id).first()

    if not doc or doc.owner_id != current_user.id:

        raise HTTPException(status_code=404)

    url = await filestack_service.thumbnail_url(doc.filestack_handle, width, height)

    return {"url": url}

The transformation runs on Filestack’s processing engine and the result is cached on the CDN. Your FastAPI app never touches the bytes.

Step 8: Verify Filestack webhooks

When you configure webhooks in the Filestack dashboard for events like upload completion or workflow finish, Filestack signs each request. Your route must verify the signature against the raw body.

# routers/webhooks.py

import json

from fastapi import APIRouter, HTTPException, Request

from fastapi.concurrency import run_in_threadpool

from filestack.helpers import verify_webhook_signature

from config import settings

router = APIRouter(prefix="/webhooks", tags=["webhooks"])


@router.post("/filestack")

async def filestack_webhook(request: Request):

    raw_body = await request.body()  # raw bytes, must come before any JSON parsing

    headers = {

        "FS-Signature": request.headers.get("FS-Signature", ""),

        "FS-Timestamp": request.headers.get("FS-Timestamp", ""),

    }

    valid, details = await run_in_threadpool(

        verify_webhook_signature,

        settings.filestack_webhook_secret,

        raw_body,

        headers,

    )

    if not valid:

        raise HTTPException(status_code=400, detail=f"Invalid signature: {details.get('error')}")

    payload = json.loads(raw_body)

    action = payload.get("action")

    # Branch on action: fp.upload, fp.video_complete, fp.workflow, etc.

    # Your business logic goes here.

    return {"received": True}

Two things that catch people out. First, read await request.body() as raw bytes before doing anything else. Reading await request.json() first will consume the stream and break signature verification. Second, verify_webhook_signature is a sync helper, so wrap it in run_in_threadpool to stay consistent with the rest of your async code.

Register the routers in your app entry point:

# app.py

from fastapi import FastAPI

from routers import uploads, webhooks

app = FastAPI(title="My API")

app.include_router(uploads.router)

app.include_router(webhooks.router)

Step 9: Background processing for large files

If you accept large files, run post-upload work like virus scanning, transcoding, or sending notifications in the background so the request returns fast.

There’s one important gotcha. FastAPI’s BackgroundTasks automatically closes the UploadFile once the response is sent, so background tasks need the bytes or handle rather than the UploadFile itself. Read the bytes inside the route, then pass those bytes (or the handle) to the task.

# routers/uploads.py (continued)

from fastapi import BackgroundTasks


async def post_upload_processing(handle: str, owner_id: int):

    """Background work: pull metadata, run AI tagging, notify the user."""

    metadata = await filestack_service.fetch_metadata(

        handle, fields=["size", "mimetype", "filename"]

    )

    # Run image tagging, transcode video, send notification email, etc.

    # All async-safe because we already converted to bytes/handles

    print(f"Processed {handle} for user {owner_id}: {metadata}")


@router.post("/with-processing", status_code=201)

async def upload_with_processing(

    background_tasks: BackgroundTasks,

    file: UploadFile = File(...),

    current_user: User = Depends(get_current_user),

):

    filelink = await filestack_service.upload_uploadfile(file, owner_id=current_user.id)

    # Schedule async background work using the handle, not the UploadFile

    background_tasks.add_task(post_upload_processing, filelink.handle, current_user.id)

    return {"handle": filelink.handle, "url": filelink.url}

For heavier workloads, swap BackgroundTasks for Celery, RQ, or Arq. The pattern stays the same: pass the handle, let the worker fetch metadata and run transformations.

Step 10: Pair with the JavaScript Picker for client-side uploads

filestack fastapi landing demo

The static demo page served by FastAPI at /.

The Filestack JS Picker after clicking the open button.

Local file selected, ready to upload.

After upload: Filestack response JSON, /api/uploads/register response, and the live CDN preview.

The Python SDK is the right tool for server-side work: webhook handling, background jobs, batch processing, admin tooling. For end-user uploads from the browser, the Filestack JavaScript Picker is faster because the file goes from the browser straight to Filestack, skipping your FastAPI app entirely.

The pattern:

Frontend fetches a signed policy from FastAPI
Frontend uses the JS Picker (initialized with that policy) to upload directly to Filestack

Picker returns a handle to your frontend

Frontend POSTs the handle to a FastAPI endpoint

FastAPI validates, persists the handle, and runs any server-side logic

A heads-up before you wire this up: if your Filestack account has security enabled (the default for production apps), the picker will fail with 403 Forbidden on /prefetch and /multipart/start unless it’s initialized with a signed policy. Generate that policy server-side and hand it to the browser through a dedicated endpoint. Also make sure the domain you’re serving the page from is whitelisted in the Filestack developer portal, or the picker will be blocked regardless of the policy.

# routers/uploads.py (continued)

@router.get("/picker-config")

async def picker_config():

    """Returns API key + signed policy so the JS Picker can upload directly."""

    sec = filestack_service.build_security(

        expiry_seconds=3600,

        calls=["pick", "read", "store", "convert"],

    )

    return {

        "apikey": settings.filestack_api_key,

        "policy": sec.policy_b64,

        "signature": sec.signature,

    }



On the browser side, fetch that config and pass it into filestack.init so every upload from the picker carries a valid signature:

<!-- static/index.html -->

<script src="//static.filestackapi.com/filestack-js/3.x.x/filestack.min.js"></script>

<script>

const cfg = await fetch("/api/uploads/picker-config").then(r => r.json());

const client = filestack.init(cfg.apikey, {

    security: { policy: cfg.policy, signature: cfg.signature },

});

const picker = client.picker({

    maxFiles: 1,

    accept: ["image/*"],

    onUploadDone: async (res) => {

      const file = res.filesUploaded[0];

      await fetch("/api/uploads/register", {

        method: "POST",

        headers: { "Content-Type": "application/json" },

        body: JSON.stringify({ handle: file.handle }),

      });

    },

});

document.getElementById("open-picker").addEventListener("click", () => picker.open());

</script>



Once the picker has the signed policy, the upload goes browser → Filestack directly, your FastAPI app only sees the handle. Server-side, the /register endpoint persists it:

# routers/uploads.py (continued)

from pydantic import BaseModel


class RegisterHandleRequest(BaseModel):

    handle: str


@router.post("/register", status_code=201)

async def register_handle(

    payload: RegisterHandleRequest,

    current_user: User = Depends(get_current_user),

    db: Session = Depends(get_db),

):

    metadata = await filestack_service.fetch_metadata(

        payload.handle, fields=["size", "mimetype", "filename"]

    )

    doc = Document(

        owner_id=current_user.id,

        filestack_handle=payload.handle,

        filestack_url=f"https://cdn.filestackcontent.com/{payload.handle}",

        original_name=metadata.get("filename", ""),

        size_bytes=metadata.get("size", 0),

    )

    db.add(doc)

    db.commit()

    db.refresh(doc)

    return {"id": doc.id}

Pydantic validates the payload, FastAPI handles dependency injection, and your service layer talks to Filestack. Large file transfers stay off your FastAPI servers and the SDK earns its keep on validation, persistence, and downstream processing.

fastapi upload

Putting it all together

You now have a working pattern: SDK installed, service module isolating Filestack calls, run_in_threadpool keeping the event loop responsive, dependency injection for testability, signed URLs for private files, transformations chained for delivery, verified webhooks for async events, and background tasks for post-upload work.

FastAPI does not need a Filestack-specific package because UploadFile.file is a standard SpooledTemporaryFile and run_in_threadpool solves the sync-SDK-in-async-app problem cleanly. Build the wrapper once and the rest of your FastAPI app talks to a normal Python module.

Ready to add reliable uploads to your FastAPI app? Get your free Filestack API key and start building secure uploads, signed URLs, transformations, and webhook-ready workflows.

Read More →