Spaces:
Runtime error
Runtime error
| """Object storage. Postgres stores keys; bytes live here. | |
| Production: Cloudflare R2 (S3 API). docker-compose: MinIO via R2_ENDPOINT_URL. Plain local dev with | |
| no R2 settings: LocalStorage, a folder on disk whose presigned URLs are HMAC signed API routes, so | |
| every caller (browser uploads, the renderer, media players) works unchanged. | |
| """ | |
| import hashlib | |
| import hmac | |
| import mimetypes | |
| import re | |
| import shutil | |
| import time | |
| import uuid | |
| from functools import cached_property | |
| from pathlib import Path | |
| from urllib.parse import quote, urlencode | |
| import boto3 | |
| from botocore.client import Config | |
| from botocore.exceptions import ClientError | |
| from app.config import settings | |
| class keys: # noqa: N801 (namespace for key builders) | |
| def raw_html(workspace_id: uuid.UUID, source_id: uuid.UUID, document_id: uuid.UUID) -> str: | |
| return f"ws/{workspace_id}/sources/{source_id}/documents/{document_id}/raw.html" | |
| def upload(workspace_id: uuid.UUID, upload_id: uuid.UUID, filename: str) -> str: | |
| return f"ws/{workspace_id}/uploads/{upload_id}/{safe_filename(filename)}" | |
| def artifact(workspace_id: uuid.UUID, artifact_id: uuid.UUID, variant: str, ext: str) -> str: | |
| return f"ws/{workspace_id}/artifacts/{artifact_id}/{variant}.{ext.lstrip('.')}" | |
| def tts_cache(workspace_id: uuid.UUID, script: str, voice_id: str, settings_json: str) -> str: | |
| digest = hashlib.sha256(f"{script}|{voice_id}|{settings_json}".encode()).hexdigest() | |
| return f"ws/{workspace_id}/tts-cache/{digest}.mp3" | |
| def dspy_program(module: str, version: str) -> str: | |
| return f"system/dspy/{module}/{version}.json" | |
| def safe_filename(name: str) -> str: | |
| name = name.rsplit("/", 1)[-1].rsplit("\\", 1)[-1] | |
| name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-.") | |
| return name[:120] or "file" | |
| def workspace_owns_key(workspace_id: uuid.UUID, key: str) -> bool: | |
| return key.startswith(f"ws/{workspace_id}/") and ".." not in key | |
| class Storage: | |
| def __init__(self, bucket: str | None = None): | |
| self.bucket = bucket or settings.r2_bucket | |
| self.prefix = settings.r2_prefix.strip("/") + "/" if settings.r2_prefix.strip("/") else "" | |
| def _k(self, key: str) -> str: | |
| return f"{self.prefix}{key}" | |
| def client(self): | |
| return boto3.client( | |
| "s3", | |
| endpoint_url=settings.r2_endpoint, | |
| aws_access_key_id=settings.r2_access_key_id, | |
| aws_secret_access_key=settings.r2_secret_access_key, | |
| region_name="auto", | |
| config=Config(signature_version="s3v4", s3={"addressing_style": "path"}), | |
| ) | |
| def ensure_bucket(self) -> None: | |
| """Local dev convenience (MinIO). On R2 create the bucket in the dashboard.""" | |
| try: | |
| self.client.head_bucket(Bucket=self.bucket) | |
| except ClientError: | |
| self.client.create_bucket(Bucket=self.bucket) | |
| def put_bytes(self, key: str, data: bytes, content_type: str | None = None) -> str: | |
| content_type = content_type or mimetypes.guess_type(key)[0] or "application/octet-stream" | |
| self.client.put_object(Bucket=self.bucket, Key=self._k(key), Body=data, ContentType=content_type) | |
| return key | |
| def put_text(self, key: str, text: str, content_type: str = "text/plain; charset=utf-8") -> str: | |
| return self.put_bytes(key, text.encode("utf-8"), content_type) | |
| def get_bytes(self, key: str) -> bytes: | |
| return self.client.get_object(Bucket=self.bucket, Key=self._k(key))["Body"].read() | |
| def download_to(self, key: str, fileobj) -> None: | |
| """Stream an object into an open binary file (large files, e.g. a video to post, never fully in memory).""" | |
| self.client.download_fileobj(self.bucket, self._k(key), fileobj) | |
| def head(self, key: str) -> dict | None: | |
| try: | |
| return self.client.head_object(Bucket=self.bucket, Key=self._k(key)) | |
| except ClientError as exc: | |
| if exc.response.get("Error", {}).get("Code") in {"404", "NoSuchKey", "NotFound"}: | |
| return None | |
| raise | |
| def exists(self, key: str) -> bool: | |
| return self.head(key) is not None | |
| def delete(self, key: str) -> None: | |
| self.client.delete_object(Bucket=self.bucket, Key=self._k(key)) | |
| def list_keys(self, prefix: str) -> list[tuple[str, float]]: | |
| """(key, last modified epoch seconds) of every object under the prefix.""" | |
| out = [] | |
| paginator = self.client.get_paginator("list_objects_v2") | |
| for page in paginator.paginate(Bucket=self.bucket, Prefix=self._k(prefix)): | |
| out += [(o["Key"][len(self.prefix):], o["LastModified"].timestamp()) for o in page.get("Contents", [])] | |
| return out | |
| def delete_prefix(self, prefix: str) -> int: | |
| deleted = 0 | |
| paginator = self.client.get_paginator("list_objects_v2") | |
| for page in paginator.paginate(Bucket=self.bucket, Prefix=self._k(prefix)): | |
| objects = [{"Key": o["Key"]} for o in page.get("Contents", [])] | |
| if objects: | |
| self.client.delete_objects(Bucket=self.bucket, Delete={"Objects": objects}) | |
| deleted += len(objects) | |
| return deleted | |
| def presign_get(self, key: str, ttl: int | None = None, download_name: str | None = None) -> str: | |
| params = {"Bucket": self.bucket, "Key": self._k(key)} | |
| if download_name: | |
| params["ResponseContentDisposition"] = f'attachment; filename="{safe_filename(download_name)}"' | |
| return self.client.generate_presigned_url( | |
| "get_object", Params=params, ExpiresIn=ttl or settings.r2_presign_ttl_seconds | |
| ) | |
| def presign_put(self, key: str, content_type: str, ttl: int | None = None) -> str: | |
| return self.client.generate_presigned_url( | |
| "put_object", | |
| Params={"Bucket": self.bucket, "Key": self._k(key), "ContentType": content_type}, | |
| ExpiresIn=ttl or settings.r2_presign_ttl_seconds, | |
| ) | |
| def public_url(self, key: str) -> str | None: | |
| if not settings.r2_public_base_url: | |
| return None | |
| return f"{settings.r2_public_base_url.rstrip('/')}/{self._k(key)}" | |
| def _local_sig(method: str, key: str, exp: int) -> str: | |
| msg = f"{method.upper()}:{key}:{exp}".encode() | |
| return hmac.new(settings.jwt_secret.encode(), msg, hashlib.sha256).hexdigest() | |
| def verify_local_sig(method: str, key: str, exp: int, sig: str) -> bool: | |
| return exp >= time.time() and hmac.compare_digest(_local_sig(method, key, exp), sig) | |
| class LocalStorage: | |
| """Disk backed stand in for R2 with the same interface.""" | |
| def __init__(self, root: str | None = None): | |
| self.root = Path(root or settings.local_storage_dir).resolve() | |
| self.bucket = "local" | |
| def path(self, key: str) -> Path: | |
| if ".." in key.split("/") or key.startswith("/"): | |
| raise ValueError(f"Invalid key: {key!r}") | |
| return self.root / key | |
| def ensure_bucket(self) -> None: | |
| self.root.mkdir(parents=True, exist_ok=True) | |
| def put_bytes(self, key: str, data: bytes, content_type: str | None = None) -> str: | |
| path = self.path(key) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(data) | |
| return key | |
| def put_text(self, key: str, text: str, content_type: str = "text/plain; charset=utf-8") -> str: | |
| return self.put_bytes(key, text.encode("utf-8"), content_type) | |
| def get_bytes(self, key: str) -> bytes: | |
| path = self.path(key) | |
| if not path.is_file(): | |
| raise FileNotFoundError(key) | |
| return path.read_bytes() | |
| def download_to(self, key: str, fileobj) -> None: | |
| path = self.path(key) | |
| if not path.is_file(): | |
| raise FileNotFoundError(key) | |
| with path.open("rb") as src: | |
| shutil.copyfileobj(src, fileobj) | |
| def head(self, key: str) -> dict | None: | |
| path = self.path(key) | |
| if not path.is_file(): | |
| return None | |
| return {"ContentLength": path.stat().st_size, "ContentType": mimetypes.guess_type(key)[0]} | |
| def exists(self, key: str) -> bool: | |
| return self.path(key).is_file() | |
| def delete(self, key: str) -> None: | |
| self.path(key).unlink(missing_ok=True) | |
| def list_keys(self, prefix: str) -> list[tuple[str, float]]: | |
| base = self.path(prefix.rstrip("/")) | |
| if not base.is_dir(): | |
| return [] | |
| return [(f"{prefix.rstrip('/')}/{f.relative_to(base).as_posix()}", f.stat().st_mtime) | |
| for f in base.rglob("*") if f.is_file()] | |
| def delete_prefix(self, prefix: str) -> int: | |
| base = self.path(prefix.rstrip("/")) | |
| if base.is_file(): | |
| base.unlink() | |
| return 1 | |
| if not base.is_dir(): | |
| return 0 | |
| count = sum(1 for f in base.rglob("*") if f.is_file()) | |
| shutil.rmtree(base, ignore_errors=True) | |
| return count | |
| def _signed(self, method: str, key: str, ttl: int | None, extra: dict | None = None) -> str: | |
| exp = int(time.time()) + (ttl or settings.r2_presign_ttl_seconds) | |
| params = {"exp": exp, "sig": _local_sig(method, key, exp), **(extra or {})} | |
| return f"{settings.api_url.rstrip('/')}/api/storage/local/{quote(key)}?{urlencode(params)}" | |
| def presign_get(self, key: str, ttl: int | None = None, download_name: str | None = None) -> str: | |
| return self._signed("GET", key, ttl, {"download": safe_filename(download_name)} if download_name else None) | |
| def presign_put(self, key: str, content_type: str, ttl: int | None = None) -> str: | |
| return self._signed("PUT", key, ttl) | |
| def public_url(self, key: str) -> str | None: | |
| return None | |
| def make_storage() -> Storage | LocalStorage: | |
| return LocalStorage() if settings.use_local_storage else Storage() | |
| storage = make_storage() | |