MinIO Storage Integration: File Upload & Management System

MinIO Storage Integration - File Upload/Management System

File management is a core requirement in an AI workflow platform. To handle the variety of files involved — RAG documents, OCR images, training data, and more — we adopted MinIO as an S3-compatible object store.

Why MinIO

• Full API compatibility with AWS S3
• Self-hosted, keeping data sovereignty in-house
• Simple deployment via Docker Compose
• Built-in web console

Docker Compose Configuration

services:
  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    volumes:
      - minio_data:/data
    command: server /data --console-address ":9001"

Python Client Implementation

from minio import Minio
from minio.error import S3Error
from fastapi import UploadFile
import io
import uuid

class StorageService:
    def __init__(self):
        self.client = Minio(
            endpoint="localhost:9000",
            access_key="minioadmin",
            secret_key="minioadmin",
            secure=False,
        )

    def ensure_bucket(self, bucket_name: str):
        if not self.client.bucket_exists(bucket_name):
            self.client.make_bucket(bucket_name)

    async def upload_file(
        self, file: UploadFile, bucket: str = "documents"
    ) -> str:
        self.ensure_bucket(bucket)
        file_ext = file.filename.split(".")[-1] if file.filename else "bin"
        object_name = f"{uuid.uuid4()}.{file_ext}"

        contents = await file.read()
        self.client.put_object(
            bucket_name=bucket,
            object_name=object_name,
            data=io.BytesIO(contents),
            length=len(contents),
            content_type=file.content_type or "application/octet-stream",
        )
        return object_name

    def get_presigned_url(
        self, object_name: str, bucket: str = "documents"
    ) -> str:
        from datetime import timedelta
        return self.client.presigned_get_object(
            bucket_name=bucket,
            object_name=object_name,
            expires=timedelta(hours=1),
        )

    def delete_file(self, object_name: str, bucket: str = "documents"):
        self.client.remove_object(bucket_name=bucket, object_name=object_name)

storage = StorageService()

FastAPI Router

from fastapi import APIRouter, UploadFile, File, Depends

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

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    object_name = await storage.upload_file(file)
    url = storage.get_presigned_url(object_name)
    return {"object_name": object_name, "url": url}

@router.get("/{object_name}/url")
async def get_file_url(object_name: str):
    url = storage.get_presigned_url(object_name)
    return {"url": url}

@router.delete("/{object_name}")
async def delete_file(object_name: str):
    storage.delete_file(object_name)
    return {"status": "deleted"}

Bucket-per-Purpose File Organization Strategy

Files are separated into buckets by use case.

MinIO's lifecycle policy feature makes automatic deletion straightforward. In production, this has significantly reduced storage costs.