74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
from abc import ABC, abstractmethod
|
|
from pathlib import Path
|
|
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class StorageBackend(ABC):
|
|
@abstractmethod
|
|
def save(self, key: str, data: bytes) -> str:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
def public_url(self, key: str) -> str:
|
|
raise NotImplementedError
|
|
|
|
|
|
class LocalStorage(StorageBackend):
|
|
def __init__(self, root: str) -> None:
|
|
self.root = Path(root)
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
|
|
def save(self, key: str, data: bytes) -> str:
|
|
path = self.root / key.replace("\\", "/").lstrip("/")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
return key.replace("\\", "/").lstrip("/")
|
|
|
|
def public_url(self, key: str) -> str:
|
|
normalized = key.replace("\\", "/").lstrip("/")
|
|
return f"/media/{normalized}"
|
|
|
|
|
|
class S3Storage(StorageBackend):
|
|
def __init__(self) -> None:
|
|
import boto3
|
|
|
|
if not settings.s3_bucket:
|
|
raise RuntimeError("S3_BUCKET is required when STORAGE_BACKEND=s3")
|
|
self.bucket = settings.s3_bucket
|
|
self.client = boto3.client(
|
|
"s3",
|
|
region_name=settings.aws_region,
|
|
aws_access_key_id=settings.aws_access_key_id,
|
|
aws_secret_access_key=settings.aws_secret_access_key,
|
|
)
|
|
base = settings.s3_public_base_url
|
|
self.public_base = (
|
|
base.rstrip("/") if base else f"https://{self.bucket}.s3.{settings.aws_region}.amazonaws.com"
|
|
)
|
|
|
|
def save(self, key: str, data: bytes) -> str:
|
|
normalized = key.replace("\\", "/").lstrip("/")
|
|
self.client.put_object(Bucket=self.bucket, Key=normalized, Body=data)
|
|
return normalized
|
|
|
|
def public_url(self, key: str) -> str:
|
|
normalized = key.replace("\\", "/").lstrip("/")
|
|
return f"{self.public_base}/{normalized}"
|
|
|
|
|
|
_storage: StorageBackend | None = None
|
|
|
|
|
|
def get_storage() -> StorageBackend:
|
|
global _storage
|
|
if _storage is None:
|
|
backend = settings.storage_backend.lower()
|
|
if backend == "s3":
|
|
_storage = S3Storage()
|
|
else:
|
|
_storage = LocalStorage(settings.upload_dir)
|
|
return _storage
|