Felipe Cardoso
9879353ac1
Renamed `base_dir` to `upload_folder` for better clarity and alignment with its purpose. Adjusted references and settings variable to reflect the updated naming convention. This enhances code readability and consistency.
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
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."""
|
|
|
|
@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}"
|
|
|
|
|
|
def get_storage_provider() -> StorageProvider:
|
|
"""Dependency for getting the configured storage provider."""
|
|
return LocalStorageProvider(
|
|
upload_folder=settings.UPLOAD_FOLDER,
|
|
files_url_path="/files"
|
|
)
|