From ba159549e28fa545b6b25149fb74bc7ef714a89b Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Mon, 15 Jun 2026 18:19:12 +0800 Subject: [PATCH] feat: integrate MinIO for real file storage --- apps/api/app/api/routes/upload.py | 42 +++++++----- apps/api/app/core/storage.py | 110 ++++++++++++++++++++++++++++++ apps/api/app/schemas/upload.py | 10 +-- requirements.txt | 2 + 4 files changed, 139 insertions(+), 25 deletions(-) create mode 100644 apps/api/app/core/storage.py diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py index 8e5ef5494..ea9f254ca 100644 --- a/apps/api/app/api/routes/upload.py +++ b/apps/api/app/api/routes/upload.py @@ -1,8 +1,10 @@ -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, UploadFile, File, Form from uuid import uuid4 +from typing import Optional from app.dependencies import get_ingest_job_repository -from app.schemas.upload import UploadAssetRequest, UploadAssetResponse +from app.core.storage import get_minio_service, MinIOService +from app.schemas.upload import UploadAssetResponse from packages.adapters.in_memory import InMemoryIngestJobRepository from packages.application import SubmitIngestJobCommand, SubmitIngestJobUseCase from apps.worker.worker_app.tasks.ingest import ingest_asset @@ -11,35 +13,42 @@ router = APIRouter() @router.post("", response_model=UploadAssetResponse) -def upload_asset( - request: UploadAssetRequest, +async def upload_asset( + file: UploadFile = File(...), + workspace_id: str = Form(...), + project_id: str = Form(...), + library_id: str = Form(...), ingest_job_repository: InMemoryIngestJobRepository = Depends(get_ingest_job_repository), + storage_service: MinIOService = Depends(get_minio_service), ) -> UploadAssetResponse: """ Upload asset and trigger ingest pipeline. - Real implementation would: + Real implementation: 1. Accept multipart/form-data with file - 2. Store file to object storage (S3, MinIO, etc.) + 2. Store file to MinIO object storage 3. Generate storage_key 4. Submit ingest job 5. Enqueue worker task - - Mock implementation: - 1. Generate mock storage_key - 2. Submit ingest job - 3. Enqueue worker task """ - # Mock storage: generate storage_key - storage_key = f"uploads/{uuid4().hex[:8]}/{request.filename}" + # Generate storage key + file_id = uuid4().hex[:8] + storage_key = f"uploads/{file_id}/{file.filename}" + + # Upload file to MinIO + file_url = storage_service.upload_file( + file.file, + storage_key, + content_type=file.content_type or "application/octet-stream", + ) # Submit ingest job use_case = SubmitIngestJobUseCase(ingest_job_repository) job = use_case.execute( SubmitIngestJobCommand( - workspace_id=request.workspace_id, - project_id=request.project_id, - library_id=request.library_id, + workspace_id=workspace_id, + project_id=project_id, + library_id=library_id, storage_key=storage_key, ) ) @@ -50,4 +59,5 @@ def upload_asset( return UploadAssetResponse( storage_key=storage_key, ingest_job_id=job.id, + url=file_url, ) diff --git a/apps/api/app/core/storage.py b/apps/api/app/core/storage.py new file mode 100644 index 000000000..fe3803d3e --- /dev/null +++ b/apps/api/app/core/storage.py @@ -0,0 +1,110 @@ +"""MinIO storage service for file uploads.""" +from minio import Minio +from minio.error import S3Error +from typing import BinaryIO +import os + + +class MinIOService: + """MinIO storage service.""" + + def __init__( + self, + endpoint: str = "47.98.113.167:9000", + access_key: str = "admin", + secret_key: str = "xiaoxia2026", + bucket_name: str = "xiaoxia-assets", + secure: bool = False, + ): + """Initialize MinIO client.""" + self.client = Minio( + endpoint, + access_key=access_key, + secret_key=secret_key, + secure=secure, + ) + self.bucket_name = bucket_name + self._ensure_bucket() + + def _ensure_bucket(self): + """Ensure bucket exists.""" + try: + if not self.client.bucket_exists(self.bucket_name): + self.client.make_bucket(self.bucket_name) + # Set download policy for public access + policy = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": ["s3:GetObject"], + "Resource": [f"arn:aws:s3:::{self.bucket_name}/*"] + } + ] + } + import json + self.client.set_bucket_policy(self.bucket_name, json.dumps(policy)) + except S3Error as e: + print(f"Error ensuring bucket: {e}") + + def upload_file( + self, + file: BinaryIO, + storage_key: str, + content_type: str = "application/octet-stream", + ) -> str: + """ + Upload file to MinIO. + + Args: + file: File object to upload + storage_key: Storage path/key (e.g., "uploads/abc123/video.mp4") + content_type: MIME type of the file + + Returns: + str: Public URL of uploaded file + """ + try: + # Get file size + file.seek(0, os.SEEK_END) + file_size = file.tell() + file.seek(0) + + # Upload file + self.client.put_object( + self.bucket_name, + storage_key, + file, + file_size, + content_type=content_type, + ) + + # Return public URL + return f"http://47.98.113.167:9000/{self.bucket_name}/{storage_key}" + + except S3Error as e: + raise Exception(f"Failed to upload file: {e}") + + def get_url(self, storage_key: str) -> str: + """Get public URL for a storage key.""" + return f"http://47.98.113.167:9000/{self.bucket_name}/{storage_key}" + + def delete_file(self, storage_key: str): + """Delete file from MinIO.""" + try: + self.client.remove_object(self.bucket_name, storage_key) + except S3Error as e: + print(f"Error deleting file: {e}") + + +# Singleton instance +_minio_service = None + + +def get_minio_service() -> MinIOService: + """Get or create MinIO service instance.""" + global _minio_service + if _minio_service is None: + _minio_service = MinIOService() + return _minio_service diff --git a/apps/api/app/schemas/upload.py b/apps/api/app/schemas/upload.py index 8e1097ad0..9dccc76c6 100644 --- a/apps/api/app/schemas/upload.py +++ b/apps/api/app/schemas/upload.py @@ -1,15 +1,7 @@ from pydantic import BaseModel, Field -class UploadAssetRequest(BaseModel): - workspace_id: str = Field(..., min_length=1) - project_id: str = Field(..., min_length=1) - library_id: str = Field(..., min_length=1) - # In real implementation: file would be UploadFile from FastAPI - # For now we'll use a mock storage_key - filename: str = Field(..., min_length=1, max_length=255) - - class UploadAssetResponse(BaseModel): storage_key: str ingest_job_id: str + url: str = Field(..., description="Public URL of uploaded file") diff --git a/requirements.txt b/requirements.txt index 49544374c..d43aa52a9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,5 @@ redis sqlalchemy alembic psycopg[binary] +minio +python-multipart