Felipe Cardoso
0cf5498762
Replaced calls to get_file_url with get_uploaded_file_url where appropriate. Added the get_uploaded_file_url method to the StorageProvider class and its implementations for more specific URL generation. Updated corresponding tests and API routes to reflect this change.
94 lines
3.0 KiB
Python
94 lines
3.0 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."""
|
|
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_uploaded_file_url(self, file_path: str) -> str:
|
|
"""Get the URL for accessing a file."""
|
|
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.upload_folder}/{file_path}"
|
|
|
|
return upload_url, file_url
|
|
|
|
def get_uploaded_file_url(self, file_path: str) -> str:
|
|
"""Get the URL for accessing a file."""
|
|
return f"{self.upload_folder}/{file_path}"
|
|
|
|
def get_file_url(self, file_path: str) -> str:
|
|
"""Get the URL for accessing a file."""
|
|
return f"{self.files_url_path}/{file_path}"
|