import enum
import datetime as dt
from sqlalchemy import Column, Integer, String, BigInteger, Enum, DateTime
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class UploadStatus(str, enum.Enum):
pending = "pending"
valid = "valid"
invalid = "invalid"
errored = "errored"
class Upload(Base):
__tablename__ = "uploads"
id = Column(Integer, primary_key=True)
original_name = Column(String(255), nullable=False)
stored_path = Column(String(512), nullable=False, unique=True)
content_type = Column(String(127), nullable=True)
size = Column(BigInteger, nullable=False, default=0)
sha256 = Column(String(64), nullable=True)
status = Column(Enum(UploadStatus), nullable=False, default=UploadStatus.pending)
reason = Column(String(512), nullable=True)
created_at = Column(DateTime, default=dt.datetime.utcnow, nullable=False)
def public_dict(self):
return {
"id": self.id,
"original_name": self.original_name,
"status": self.status.value,
"size": self.size,
"reason": self.reason,
}
import hashlib
import uuid
from pathlib import Path
from fastapi import APIRouter, UploadFile, File, BackgroundTasks, Depends, status
from sqlalchemy.orm import Session
from .models import Upload, UploadStatus
from .db import get_db
from .validation import validate_upload
router = APIRouter(prefix="/uploads", tags=["uploads"])
UPLOAD_DIR = Path("var/uploads")
CHUNK_SIZE = 1024 * 1024 # 1 MiB
@router.post("", status_code=status.HTTP_202_ACCEPTED)
async def create_upload(
background_tasks: BackgroundTasks,
file: UploadFile = File(...),
db: Session = Depends(get_db),
):
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
dest = UPLOAD_DIR / f"{uuid.uuid4().hex}{Path(file.filename).suffix}"
digest = hashlib.sha256()
size = 0
with dest.open("wb") as out:
while chunk := await file.read(CHUNK_SIZE):
digest.update(chunk)
size += len(chunk)
out.write(chunk)
await file.close()
record = Upload(
original_name=file.filename,
stored_path=str(dest),
content_type=file.content_type,
size=size,
sha256=digest.hexdigest(),
status=UploadStatus.pending,
)
db.add(record)
db.commit()
db.refresh(record)
background_tasks.add_task(validate_upload, record.id)
return record.public_dict()
@router.get("/{upload_id}")
def get_upload(upload_id: int, db: Session = Depends(get_db)):
row = db.query(Upload).get(upload_id)
if row is None:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="not found")
return row.public_dict()
import logging
from pathlib import Path
import magic
from .db import SessionLocal
from .models import Upload, UploadStatus
logger = logging.getLogger("uploads.validation")
MAX_BYTES = 200 * 1024 * 1024
ALLOWED_TYPES = {"image/png", "image/jpeg", "application/pdf"}
def validate_upload(upload_id: int) -> None:
db = SessionLocal()
try:
row = db.query(Upload).get(upload_id)
if row is None or row.status != UploadStatus.pending:
return
path = Path(row.stored_path)
if not path.exists():
_finish(db, row, UploadStatus.errored, "stored file missing")
return
if row.size > MAX_BYTES:
_finish(db, row, UploadStatus.invalid, "file exceeds size limit")
return
sniffed = magic.from_file(str(path), mime=True)
if sniffed not in ALLOWED_TYPES:
_finish(db, row, UploadStatus.invalid, f"disallowed type: {sniffed}")
return
row.content_type = sniffed
_finish(db, row, UploadStatus.valid, None)
except Exception as exc: # noqa: BLE001
logger.exception("validation failed for upload %s", upload_id)
try:
row = db.query(Upload).get(upload_id)
if row is not None:
_finish(db, row, UploadStatus.errored, str(exc)[:500])
except Exception:
db.rollback()
finally:
db.close()
def _finish(db, row, status, reason):
row.status = status
row.reason = reason
db.add(row)
db.commit()
This snippet shows how FastAPI accepts a file upload, streams it to disk without exhausting memory, records a pending row, and then hands off content validation to a BackgroundTask so the request returns quickly.
The route in uploads router uses UploadFile = File(...), which gives a SpooledTemporaryFile under the hood: small files stay in memory, larger ones spill to disk. Rather than calling await file.read() (which pulls the entire body into RAM), the handler reads in fixed chunks with await file.read(CHUNK_SIZE) and writes them out, so a 500MB upload costs a few KB of working memory. A SHA-256 hash is accumulated during the same pass, giving a content fingerprint for free. The saved path uses a random uuid4 name to avoid collisions and path-traversal from user-controlled filenames.
Before returning, the handler inserts an Upload row in state pending and schedules validate_upload via background_tasks.add_task(...). FastAPI runs that callable after the response is sent, so the client gets an immediate 202 Accepted while heavier work continues out of band. This is the key trade-off: validation is eventually-consistent, so clients must poll or subscribe for the final status instead of expecting it inline.
validation task opens a fresh database session — background tasks run outside the request scope, so the request-bound session is already closed. It re-reads the file to check the real content type with python-magic, enforces a size ceiling, and flips the row to valid or invalid, storing a reason. Wrapping the body in try/except and marking a row errored matters because an unhandled exception in a background task is logged but invisible to the user.
The Upload model defines the durable state machine via an UploadStatus enum and stores sha256, size, and content_type. Because validation is idempotent on the stored path, a failed task can be safely retried. For truly durable processing across restarts a real queue like Celery or RQ is preferable, but BackgroundTask is the right lightweight tool when the work is short and best-effort.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
class SignupForm
include ActiveModel::Model
include ActiveModel::Attributes
attribute :account_name, :string
attribute :email, :stringShallow Controller, Deep Params: Form Object Pattern
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Form Validation Example</title>
<style>HTML forms with validation and accessibility
Share this code
Here's the card — post it anywhere.
