Felipe Cardoso
ffdf095eee
Updated file URL generation to return relative paths instead of absolute paths, ensuring better portability and consistency. Adjusted relevant methods to compute paths relative to the base directory.
96 lines
3.2 KiB
Python
96 lines
3.2 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}"
|
|
|
|
# Return just the relative path
|
|
upload_folder_relative = os.path.relpath(self.upload_folder, settings.DATA_FILES_DIR)
|
|
file_url = f"{upload_folder_relative}/{file_path}"
|
|
|
|
return upload_url, file_url
|
|
|
|
def get_uploaded_file_url(self, file_path: str) -> str:
|
|
"""Get the URL for accessing a file."""
|
|
upload_folder_relative = os.path.relpath(self.upload_folder, settings.DATA_FILES_DIR)
|
|
return f"{upload_folder_relative}/{file_path}"
|
|
|
|
def get_file_url(self, file_path: str) -> str:
|
|
"""Get the URL for accessing a file."""
|
|
# Return just the relative path
|
|
return file_path |