181 lines
5.4 KiB
Python
181 lines
5.4 KiB
Python
"""Image storage abstraction: local filesystem or S3-compatible object storage.
|
|
|
|
S3 config is read from ``s3_config.json`` in the project root::
|
|
|
|
{
|
|
"enabled": true,
|
|
"endpoint": "https://s3.amazonaws.com",
|
|
"bucket": "my-mailova-bucket",
|
|
"region": "auto",
|
|
"access_key": "AKIA...",
|
|
"secret_key": "...",
|
|
"public_url": "https://cdn.example.com/mailova"
|
|
}
|
|
|
|
When the file is missing or ``enabled`` is false, local storage is used.
|
|
|
|
URL scheme
|
|
~~~~~~~~~~
|
|
Database stores **bare filenames** (``xxx.jpeg``) regardless of backend.
|
|
``resolve_url(key)`` converts a bare key to a full URL:
|
|
- S3 → ``<public_url>/xxx.jpeg``
|
|
- local → ``/uploads/xxx.jpeg``
|
|
|
|
Legacy records that still contain ``/uploads/xxx.jpeg`` are auto-stripped
|
|
by ``resolve_url``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
|
|
from app.config import UPLOAD_DIR
|
|
|
|
S3_CONFIG_PATH = Path(__file__).resolve().parent.parent / "s3_config.json"
|
|
|
|
|
|
class StorageBackend(Protocol):
|
|
def put(self, key: str, data: bytes, content_type: str = "image/jpeg") -> str:
|
|
"""Store data and return the bare key."""
|
|
...
|
|
|
|
def delete(self, key: str) -> None: ...
|
|
|
|
def exists(self, key: str) -> bool: ...
|
|
|
|
def resolve_url(self, key: str) -> str:
|
|
"""Convert a bare key (or legacy /uploads/xxx path) to a full URL."""
|
|
...
|
|
|
|
|
|
# ── Local Storage ──────────────────────────────────────────────────
|
|
|
|
class LocalStorage:
|
|
def __init__(self) -> None:
|
|
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
def put(self, key: str, data: bytes, content_type: str = "") -> str:
|
|
(UPLOAD_DIR / key).write_bytes(data)
|
|
return key
|
|
|
|
def delete(self, key: str) -> None:
|
|
p = UPLOAD_DIR / key
|
|
if p.exists():
|
|
p.unlink()
|
|
|
|
def exists(self, key: str) -> bool:
|
|
return (UPLOAD_DIR / key).exists()
|
|
|
|
def resolve_url(self, key: str) -> str:
|
|
return f"/uploads/{_strip_prefix(key)}"
|
|
|
|
|
|
# ── S3 Storage ─────────────────────────────────────────────────────
|
|
|
|
class S3Storage:
|
|
def __init__(
|
|
self,
|
|
endpoint_url: str,
|
|
bucket: str,
|
|
access_key: str,
|
|
secret_key: str,
|
|
region: str = "auto",
|
|
public_url: str = "",
|
|
) -> None:
|
|
import boto3
|
|
|
|
self.bucket = bucket
|
|
self.public_url = public_url.rstrip("/")
|
|
self._client = boto3.client(
|
|
"s3",
|
|
endpoint_url=endpoint_url or None,
|
|
aws_access_key_id=access_key,
|
|
aws_secret_access_key=secret_key,
|
|
region_name=region or "auto",
|
|
)
|
|
|
|
def put(self, key: str, data: bytes, content_type: str = "image/jpeg") -> str:
|
|
self._client.put_object(
|
|
Bucket=self.bucket, Key=key, Body=data, ContentType=content_type,
|
|
)
|
|
return key
|
|
|
|
def delete(self, key: str) -> None:
|
|
self._client.delete_object(Bucket=self.bucket, Key=key)
|
|
|
|
def exists(self, key: str) -> bool:
|
|
try:
|
|
self._client.head_object(Bucket=self.bucket, Key=key)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def resolve_url(self, key: str) -> str:
|
|
k = _strip_prefix(key)
|
|
if self.public_url:
|
|
return f"{self.public_url}/{k}"
|
|
return k
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────
|
|
|
|
def _strip_prefix(key: str) -> str:
|
|
"""Strip ``/uploads/`` prefix from legacy DB values."""
|
|
if key.startswith("/uploads/"):
|
|
return key[len("/uploads/"):]
|
|
return key
|
|
|
|
|
|
# ── Singleton accessor ─────────────────────────────────────────────
|
|
|
|
_backend: StorageBackend | None = None
|
|
|
|
|
|
def _load_s3_config() -> dict | None:
|
|
"""Read s3_config.json. Returns None if not configured / disabled."""
|
|
if not S3_CONFIG_PATH.exists():
|
|
return None
|
|
try:
|
|
cfg = json.loads(S3_CONFIG_PATH.read_text(encoding="utf-8"))
|
|
if not cfg.get("enabled") or not cfg.get("bucket"):
|
|
return None
|
|
return cfg
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_storage() -> StorageBackend:
|
|
"""Get the active storage backend (S3 if configured, otherwise local)."""
|
|
global _backend
|
|
if _backend is not None:
|
|
return _backend
|
|
|
|
s3_cfg = _load_s3_config()
|
|
if s3_cfg:
|
|
_backend = S3Storage(
|
|
endpoint_url=s3_cfg.get("endpoint", ""),
|
|
bucket=s3_cfg["bucket"],
|
|
access_key=s3_cfg.get("access_key", ""),
|
|
secret_key=s3_cfg.get("secret_key", ""),
|
|
region=s3_cfg.get("region", "auto"),
|
|
public_url=s3_cfg.get("public_url", ""),
|
|
)
|
|
else:
|
|
_backend = LocalStorage()
|
|
return _backend
|
|
|
|
|
|
def resolve_url(key: str | None) -> str:
|
|
"""Convenience: resolve a possibly-None image key to a URL (or empty)."""
|
|
if not key:
|
|
return ""
|
|
return get_storage().resolve_url(key)
|
|
|
|
|
|
def reset_storage() -> None:
|
|
"""Force re-detection on next get_storage() call."""
|
|
global _backend
|
|
_backend = None
|