File size: 3,034 Bytes
5dccc28 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | from pathlib import Path
from app.config import get_settings
class StorageService:
def __init__(self) -> None:
self.settings = get_settings()
self.backend = self.settings.storage_backend.lower().strip()
self._client = None
self.settings.upload_path.mkdir(parents=True, exist_ok=True)
def save_pdf(self, *, file_hash: str, filename: str, content: bytes) -> str:
if self._use_supabase():
key = self._build_supabase_key(file_hash=file_hash, filename=filename)
self._supabase().storage.from_(self.settings.supabase_storage_bucket).upload(
path=key,
file=content,
file_options={"content-type": "application/pdf", "upsert": "true"},
)
return f"supabase://{self.settings.supabase_storage_bucket}/{key}"
suffix = Path(filename).suffix or ".pdf"
target = self.settings.upload_path / f"{file_hash}{suffix}"
if not target.exists():
target.write_bytes(content)
return str(target)
def read_file_bytes(self, *, file_path: str) -> bytes:
if file_path.startswith("supabase://"):
bucket, key = self._parse_supabase_path(file_path)
result = self._supabase().storage.from_(bucket).download(key)
if isinstance(result, bytes):
return result
return bytes(result)
return Path(file_path).read_bytes()
def delete_file(self, *, file_path: str) -> None:
if file_path.startswith("supabase://"):
bucket, key = self._parse_supabase_path(file_path)
try:
self._supabase().storage.from_(bucket).remove([key])
except Exception:
pass
return
local_path = Path(file_path)
if local_path.exists():
try:
local_path.unlink()
except OSError:
pass
def _use_supabase(self) -> bool:
return (
self.backend == "supabase"
and bool(self.settings.supabase_url)
and bool(self.settings.supabase_service_role_key)
)
def _supabase(self):
if self._client is None:
from supabase import create_client
self._client = create_client(self.settings.supabase_url, self.settings.supabase_service_role_key)
return self._client
def _build_supabase_key(self, *, file_hash: str, filename: str) -> str:
suffix = Path(filename).suffix or ".pdf"
base_prefix = self.settings.supabase_storage_prefix.strip("/ ")
if base_prefix:
return f"{base_prefix}/{file_hash}{suffix}"
return f"{file_hash}{suffix}"
def _parse_supabase_path(self, file_path: str) -> tuple[str, str]:
without_scheme = file_path.removeprefix("supabase://")
parts = without_scheme.split("/", 1)
if len(parts) != 2:
raise ValueError(f"Invalid supabase path: {file_path}")
return parts[0], parts[1]
|