KAHIBARO
Discord Login Register ☰
☰

File Storage

Table of Contents

Why File Storage Matters in a Production Backend

In a production application, files are not just an afterthought. User avatars, invoices, logs, exported reports, product images, documents, and media all live somewhere. How you store them affects:

In the final project, you already have APIs, databases, authentication, and background jobs. File storage must integrate smoothly with all of these and work reliably under real-world conditions.

This chapter focuses on choosing and implementing a file storage strategy for your production-ready project, not re-explaining basic upload APIs.

Core Concepts: Files vs Database

A common beginner mistake is to store binary files directly in the database. In a production backend you usually separate:

Typical metadata table:

ColumnDescription
idPrimary key
owner_user_idForeign key to users
original_nameName provided by client
stored_pathPath or key in your storage system
mime_typeimage/png, application/pdf, etc.
size_bytesFile size
checksumHash like SHA-256 or MD5 for integrity
created_atUpload time
visibilityprivate or public

This split has several advantages:

Rule:
Store file metadata in the database and file contents in a dedicated storage system (file system or object storage) in production systems.

Storage Options for the Final Project

You will likely support at least two storage backends:

  1. Local file system
    • Simple, good for development and small deployments.
    • Files live in something like ./storage/ inside or mounted to your container.
  2. Object storage (S3 compatible)
    • Used in real production (AWS S3, MinIO, DigitalOcean Spaces, etc).
    • Good durability, scalability, and integration with CDNs.

You should design your project so you can switch using configuration:

text 
STORAGE_BACKEND=local             # or: s3
LOCAL_STORAGE_PATH=/app/storage
S3_BUCKET_NAME=myapp-files
S3_REGION=eu-central-1
S3_ENDPOINT_URL=https://s3.amazonaws.com  # or custom for MinIO

Designing a File Storage Abstraction

To keep your application clean, do not call S3 or the filesystem directly from your route handlers. Instead, define a small interface that your application uses everywhere.

For example, you can define a FileStorage protocol or base class:

python 
from typing import Protocol, BinaryIO, Optional
class FileStorage(Protocol):
    async def save(
        self,
        path: str,
        file_obj: BinaryIO,
        content_type: Optional[str] = None,
    ) -> None:
        ...
    async def open(self, path: str) -> BinaryIO:
        ...
    async def delete(self, path: str) -> None:
        ...
    async def generate_public_url(self, path: str) -> str:
        ...
    async def generate_presigned_download_url(
        self,
        path: str,
        expires_in_seconds: int = 3600,
    ) -> str:
        ...

Then implement at least two concrete classes:

Route handlers depend only on FileStorage, so switching implementation is easy through dependency injection.

Rule:
Always wrap your storage provider behind an application-level interface instead of sprinkling S3 or filesystem calls throughout your code.

Local File Storage in a Production Context

Path Structure and Naming

Your application should not trust client file names. Instead, generate safe paths:

Example internal path pattern:

text 
{environment}/{resource_type}/{year}/{month}/{uuid4}.{extension}

Example concrete path:

text 
prod/user-avatars/2026/08/18b141e6-45e2-4f0c-b17c-fc3cf5d7a0b2.png

You can map that to a local directory:

text 
/app/storage/prod/user-avatars/2026/08/18b141e6-45e2-4f0c-b17c-fc3cf5d7a0b2.png

Implementing Local Storage

A minimal async friendly implementation with blocking I/O wrapped into a thread:

python 
import os
from pathlib import Path
from typing import BinaryIO, Optional
from fastapi.concurrency import run_in_threadpool
class LocalFileStorage:
    def __init__(self, base_path: str, public_base_url: str | None = None):
        self.base_path = Path(base_path)
        self.public_base_url = public_base_url
    async def save(
        self,
        path: str,
        file_obj: BinaryIO,
        content_type: Optional[str] = None,
    ) -> None:
        full_path = self.base_path / path
        full_path.parent.mkdir(parents=True, exist_ok=True)
        async def _write():
            with open(full_path, "wb") as f:
                while True:
                    chunk = file_obj.read(1024 * 1024)
                    if not chunk:
                        break
                    f.write(chunk)
        await run_in_threadpool(_write)
    async def open(self, path: str) -> BinaryIO:
        full_path = self.base_path / path
        return open(full_path, "rb")
    async def delete(self, path: str) -> None:
        full_path = self.base_path / path
        def _delete():
            if full_path.exists():
                full_path.unlink()
        await run_in_threadpool(_delete)
    async def generate_public_url(self, path: str) -> str:
        if not self.public_base_url:
            raise RuntimeError("Public base URL not configured")
        return f"{self.public_base_url.rstrip('/')}/{path}"
    async def generate_presigned_download_url(
        self,
        path: str,
        expires_in_seconds: int = 3600,
    ) -> str:
        # For local storage, this might simply be the same as generate_public_url,
        # or a signed route in your FastAPI app.
        return await self.generate_public_url(path)

In development, public_base_url can be http://localhost:8000/files. In production, it might be an Nginx route or CDN domain.

Serving Local Files

In production you usually:

Example Nginx config snippet:

nginx 
location /files/ {
    alias /data/myapp/storage/prod/;
    autoindex off;
}

If stored_path is prod/user-avatars/2026/08/uuid.png, the public URL can be:

text 
https://cdn.myapp.com/files/user-avatars/2026/08/uuid.png

Your generate_public_url implementation must match this mapping.

Object Storage and S3-Compatible Backends

For production deployments you often use object storage instead of local disks. Benefits:

Conceptual Mapping

ConceptLocal FSS3 / Object Storage
Path/app/storage/...S3 object key
DirectoryReal directory on diskPrefix in object keys
URL/files/...https://bucket.s3.../key
Move/renamemv on file systemCopy then delete
Access controlFile permissionsBucket policies, ACLs

In your code, treat the S3 key as the same stored_path string you used for the local backend.

Implementing S3 Storage

Using aioboto3 for async S3 operations:

python 
from typing import BinaryIO, Optional
import aioboto3
class S3FileStorage:
    def __init__(
        self,
        bucket_name: str,
        region_name: str,
        endpoint_url: str | None = None,
    ):
        self.bucket_name = bucket_name
        self.region_name = region_name
        self.endpoint_url = endpoint_url
    def _session(self):
        return aioboto3.Session()
    async def save(
        self,
        path: str,
        file_obj: BinaryIO,
        content_type: Optional[str] = None,
    ) -> None:
        extra_args = {}
        if content_type:
            extra_args["ContentType"] = content_type
        async with self._session().client(
            "s3",
            region_name=self.region_name,
            endpoint_url=self.endpoint_url,
        ) as s3:
            await s3.upload_fileobj(
                Fileobj=file_obj,
                Bucket=self.bucket_name,
                Key=path,
                ExtraArgs=extra_args,
            )
    async def open(self, path: str) -> BinaryIO:
        # In many cases you will not open via the backend
        # but use presigned URLs instead.
        raise NotImplementedError("Use presigned URLs for S3")
    async def delete(self, path: str) -> None:
        async with self._session().client(
            "s3",
            region_name=self.region_name,
            endpoint_url=self.endpoint_url,
        ) as s3:
            await s3.delete_object(Bucket=self.bucket_name, Key=path)
    async def generate_public_url(self, path: str) -> str:
        # Only valid if the object is publicly accessible
        endpoint = self.endpoint_url or f"https://{self.bucket_name}.s3.{self.region_name}.amazonaws.com"
        return f"{endpoint.rstrip('/')}/{path}"
    async def generate_presigned_download_url(
        self,
        path: str,
        expires_in_seconds: int = 3600,
    ) -> str:
        async with self._session().client(
            "s3",
            region_name=self.region_name,
            endpoint_url=self.endpoint_url,
        ) as s3:
            return await s3.generate_presigned_url(
                "get_object",
                Params={"Bucket": self.bucket_name, "Key": path},
                ExpiresIn=expires_in_seconds,
            )

You will also need to configure AWS credentials via environment variables or IAM roles in production.

Generating File Paths in Your Domain Layer

To keep path logic consistent, put it in a dedicated helper or service, not in route handlers.

Example:

python 
import uuid
from datetime import datetime
from pathlib import Path
def build_storage_path(
    environment: str,
    resource_type: str,
    original_filename: str,
    now: datetime | None = None,
) -> str:
    now = now or datetime.utcnow()
    ext = Path(original_filename).suffix.lower()  # includes dot, e.g. ".png"
    unique_id = uuid.uuid4()
    return f"{environment}/{resource_type}/{now.year:04d}/{now.month:02d}/{unique_id}{ext}"

Usage in an avatar upload endpoint:

python 
from fastapi import UploadFile, Depends
from datetime import datetime
async def upload_avatar(
    file: UploadFile,
    storage: FileStorage = Depends(get_file_storage),
):
    path = build_storage_path(
        environment=settings.ENVIRONMENT,
        resource_type="user-avatars",
        original_filename=file.filename,
        now=datetime.utcnow(),
    )
    await storage.save(path, file.file, content_type=file.content_type)
    # Store metadata in DB
    avatar = await avatar_repo.create(
        user_id=current_user.id,
        stored_path=path,
        original_name=file.filename,
        mime_type=file.content_type,
        size_bytes=file.size,  # or read length explicitly
    )
    return {"id": avatar.id}

Private vs Public Files

In the final project you will likely have:

The difference is how you allow clients to access them.

Public Files

Options:

  1. Public S3 bucket or CDN.
  2. Local storage behind Nginx, with no authentication.

Your generate_public_url simply returns the final URL.

Private Files

Private files should never be world-readable. Two common patterns:

  1. Authenticated download route
    • Client requests /files/{file_id} with authentication.
    • Backend checks authorization and either streams from local disk or proxies from S3.
  2. Short-lived presigned URLs
    • Client asks your backend for a presigned URL.
    • Backend checks authorization and generates a time-limited URL directly from S3.
    • Client downloads directly from S3 using this URL.

Presigned URLs offload traffic from your app server and can scale independently.

Rule:
Store private files in private storage. Use authenticated routes or presigned URLs. Never make sensitive buckets or directories publicly readable.

Upload, Download, and Background Processing

In a production backend uploads and downloads are often combined with background jobs.

Typical Flow: Image Upload

  1. Client uploads an image.
  2. Backend validates type and size, immediately creates metadata record and saves original file.
  3. Backend enqueues a background job to:
    • Generate thumbnails
    • Optimize image size
    • Maybe extract metadata (dimensions, etc.)
  4. Background worker creates additional files in storage and updates metadata in DB.

This approach keeps the upload endpoint fast and responsive.

Example: FastAPI Upload Endpoint Using Abstraction

python 
from fastapi import APIRouter, UploadFile, File, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
router = APIRouter()
@router.post("/files/images", status_code=status.HTTP_201_CREATED)
async def upload_image(
    image: UploadFile = File(...),
    storage: FileStorage = Depends(get_file_storage),
    db: AsyncSession = Depends(get_db_session),
    current_user: User = Depends(get_current_user),
):
    if image.content_type not in {"image/png", "image/jpeg"}:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Unsupported image type",
        )
    # Limit example: no more than 5 MB
    max_size_bytes = 5 * 1024 * 1024
    content = await image.read()
    if len(content) > max_size_bytes:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="File too large",
        )
    from io import BytesIO
    file_obj = BytesIO(content)
    path = build_storage_path(
        environment=settings.ENVIRONMENT,
        resource_type="images",
        original_filename=image.filename,
    )
    await storage.save(path, file_obj, content_type=image.content_type)
    # Save metadata in DB
    db_file = await file_repo.create(
        db,
        owner_user_id=current_user.id,
        original_name=image.filename,
        stored_path=path,
        mime_type=image.content_type,
        size_bytes=len(content),
        visibility="private",
    )
    # Enqueue background job (for example via Celery) to process image
    process_image_async.delay(file_id=db_file.id)
    return {"id": db_file.id}

In the background worker, you look up the file by id, read from storage, generate thumbnails, and create additional records.

Security Considerations for File Storage

File storage easily becomes a security hole if you are not careful.

Validation

At upload time, enforce:

Path Safety

Never accept client-provided paths or allow ../ sequences to influence your storage path. All storage paths must be generated by the server.

Rule:
Never directly use a client-provided path or file name as your storage path. Always generate a safe, controlled path on the server side.

Sanitizing File Names

If you need a user-visible file name (for example Content-Disposition: attachment; filename="..."), sanitize it:

Access Control

Tie every file to an owner or resource in your database and always check:

Virus and Malware Scanning

For real-world systems that allow arbitrary uploads, you should consider:

Backups and Disaster Recovery for Files

In your final project you will plan backup and disaster recovery. Files are part of that.

Local Storage Backups

If you use local storage in production:

Object Storage Durability

S3 and similar systems usually have built in redundancy:

Consistency Between DB and Files

You can get into states like:

Handling this in production:

Logging, Observability, and Quotas

Files can consume a lot of storage and cost. In production you often need visibility and control.

Logging

For each upload, log:

For download access, especially for private files, log:

Metrics

Collect metrics such as:

You can compute these from your DB or from your storage system and export to Prometheus.

Quotas

To prevent a single user from filling your storage:

Making Your Final Project Production Ready

For the final project, a practical way to implement file storage with production concerns:

  1. Define a FileStorage interface with methods for save, delete, and URL generation.
  2. Implement LocalFileStorage and use it in development and tests.
  3. Implement S3FileStorage for production with environment based configuration.
  4. Create a files table in PostgreSQL that stores metadata and links files to users or domain entities.
  5. Use background workers to process heavy file related tasks, such as thumbnails or PDF generation.
  6. Enforce validation, access control, and logging from the start.
  7. Integrate storage into your backup and recovery plan, treating it as critical data.

If you design this layer cleanly, you will be able to:

Views: 7

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3096 / 24h
Impressum / Imprint
Datenschutz / Data Policy