Add local storage provider implementation
Introduce an abstract `StorageProvider` class and a
This commit is contained in:
91
backend/app/core/storage.py
Normal file
91
backend/app/core/storage.py
Normal file
@@ -0,0 +1,91 @@
|
|||||||
|
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, base_dir: str, files_url_path: str):
|
||||||
|
self.base_dir = Path(base_dir)
|
||||||
|
self.files_url_path = files_url_path
|
||||||
|
|
||||||
|
# Ensure base directory exists
|
||||||
|
os.makedirs(self.base_dir, 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.base_dir / 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(
|
||||||
|
base_dir=settings.DATA_FILES_DIR,
|
||||||
|
files_url_path="/files"
|
||||||
|
)
|
||||||
Reference in New Issue
Block a user