import os import shutil from abc import ABC, abstractmethod from pathlib import Path from typing import Tuple from fastapi import UploadFile from app.core.config import settings class StorageProvider(ABC): """Base abstract class for storage providers.""" upload_folder: Path @abstractmethod async def save_file(self, file: UploadFile, destination: str) -> str: """Save a file to storage and return the relative path.""" pass @abstractmethod def generate_presigned_url(self, file_path: str, filename: str, content_type: str, expires_in: int = 300) -> Tuple[str, str]: """ Generate a presigned URL for file upload. Returns: (upload_url, file_url) """ pass @abstractmethod def get_file_url(self, file_path: str) -> str: """Get the URL for accessing a file.""" pass class LocalStorageProvider(StorageProvider): """Local filesystem storage provider.""" def __init__(self, upload_folder: str, files_url_path: str): self.upload_folder = Path(upload_folder) self.files_url_path = files_url_path # Ensure upload directory exists os.makedirs(self.upload_folder, exist_ok=True) async def save_file(self, file: UploadFile, destination: str) -> str: """Save an uploaded file to the local filesystem.""" # Ensure destination directory exists dest_path = self.upload_folder / destination os.makedirs(dest_path.parent, exist_ok=True) # Save the file with open(dest_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) # Return the relative path return destination def generate_presigned_url(self, file_path: str, filename: str, content_type: str, expires_in: int = 300) -> Tuple[str, str]: """ Generate a token-based upload URL for local storage. This simulates presigned URLs for local storage. """ from app.utils.security import create_upload_token # Generate a unique token for this upload token = create_upload_token(file_path, content_type, expires_in) # The upload URL is to our custom upload endpoint upload_url = f"{settings.API_VERSION_STR}/uploads/{token}" # The file URL is where the file will be accessible after upload file_url = f"{self.files_url_path}/{file_path}" return upload_url, file_url def get_file_url(self, file_path: str) -> str: """Get the URL for accessing a file.""" return f"{self.files_url_path}/{file_path}"