Image Processing Pipeline

Use MemoryEngine to receive images in memory and process them without touching the filesystem.

Full Example

app.py

import hashlib

from fastapi import Depends, FastAPI
from filestore import FileStore, MemoryEngine, Store, StoreConfig, UploadContext

app = FastAPI()


async def image_filter(ctx: UploadContext):
    """Verify the file starts with a known image magic number."""
    await ctx.file.seek(0)
    header = await ctx.file.read(8)
    await ctx.file.seek(0)

    png_magic = b"\x89PNG\r\n\x1a\n"
    jpeg_magic = b"\xff\xd8\xff"

    if header.startswith(png_magic) or header.startswith(jpeg_magic):
        return True
    return "File does not appear to be a valid PNG or JPEG image"


storage = FileStore(
    "image",
    required=True,
    engine=MemoryEngine(),
    config=StoreConfig(
        allowed_content_types=["image/jpeg", "image/png"],
        max_file_size=10 * 1024 * 1024,  # 10 MB
        filters=[image_filter],
    ),
)


@app.post("/images/process")
async def process_image(store: Store = Depends(storage)):
    image = store.first("image")

    if not image or not image.status:
        return {"error": store.error}

    raw_bytes = image.file  # bytes — the full image payload

    # Compute a content hash
    content_hash = hashlib.sha256(raw_bytes).hexdigest()

    return {
        "filename": image.original_filename,
        "content_type": image.content_type,
        "size": image.size,
        "sha256": content_hash,
    }

Key Points

• MemoryEngine — no disk I/O, the full payload is in image.file
• Magic number filter — validates actual file content, not just headers
• Content hashing — process bytes directly in-memory
• The raw bytes are never serialized — returning the Store from a route is safe
• Useful for thumbnailing, OCR, virus scanning, or forwarding to another service