"""Abstract storage backend — supports local filesystem and S3-compatible storage. Configuration via environment variables: - STORAGE_BACKEND: "local" (default) or "s3" - STORAGE_PATH: Local storage base path (default: /data/uploads) - S3_ENDPOINT: S3-compatible endpoint URL - S3_BUCKET: Bucket name - S3_ACCESS_KEY: Access key - S3_SECRET_KEY: Secret key - S3_REGION: Region (default: us-east-1) - S3_SECURE: Use HTTPS (default: true) """ from __future__ import annotations import io import logging import os from abc import ABC, abstractmethod from typing import Any logger = logging.getLogger(__name__) class StorageBackend(ABC): """Abstract storage backend for file operations.""" @abstractmethod async def save(self, path: str, data: bytes) -> str: """Save data to storage at the given path. Returns the full storage path.""" ... @abstractmethod async def read(self, path: str) -> bytes: """Read data from storage at the given path.""" ... @abstractmethod async def delete(self, path: str) -> bool: """Delete a file from storage. Returns True if deleted, False if not found.""" ... @abstractmethod async def exists(self, path: str) -> bool: """Check if a file exists in storage.""" ... @abstractmethod async def get_url(self, path: str, expires: int = 3600) -> str: """Get a URL for accessing the file (presigned URL for S3, file path for local).""" ... @abstractmethod async def list_files(self, prefix: str) -> list[str]: """List all file paths under the given prefix.""" ... class LocalStorage(StorageBackend): """Local filesystem storage backend.""" def __init__(self, base_path: str | None = None) -> None: self.base_path = base_path or os.environ.get("STORAGE_PATH", "/data/uploads") os.makedirs(self.base_path, exist_ok=True) def _full_path(self, path: str) -> str: """Get the full filesystem path.""" return os.path.join(self.base_path, path) async def save(self, path: str, data: bytes) -> str: full_path = self._full_path(path) os.makedirs(os.path.dirname(full_path), exist_ok=True) with open(full_path, "wb") as f: f.write(data) logger.debug("LocalStorage: saved %s (%d bytes)", path, len(data)) return path async def read(self, path: str) -> bytes: full_path = self._full_path(path) with open(full_path, "rb") as f: return f.read() async def delete(self, path: str) -> bool: full_path = self._full_path(path) if os.path.exists(full_path): os.remove(full_path) return True return False async def exists(self, path: str) -> bool: return os.path.exists(self._full_path(path)) async def get_url(self, path: str, expires: int = 3600) -> str: # Local storage returns the file path for direct access return self._full_path(path) async def list_files(self, prefix: str) -> list[str]: full_prefix = self._full_path(prefix) if not os.path.isdir(full_prefix): return [] result: list[str] = [] for root, _dirs, files in os.walk(full_prefix): for fname in files: rel = os.path.relpath(os.path.join(root, fname), self.base_path) result.append(rel) return result class S3Storage(StorageBackend): """S3-compatible storage backend (works with AWS S3, MinIO, etc.).""" def __init__( self, endpoint: str | None = None, bucket: str | None = None, access_key: str | None = None, secret_key: str | None = None, region: str | None = None, secure: bool | None = None, ) -> None: self.endpoint = endpoint or os.environ.get("S3_ENDPOINT", "") self.bucket = bucket or os.environ.get("S3_BUCKET", "") self.access_key = access_key or os.environ.get("S3_ACCESS_KEY", "") self.secret_key = secret_key or os.environ.get("S3_SECRET_KEY", "") self.region = region or os.environ.get("S3_REGION", "us-east-1") self.secure = secure if secure is not None else os.environ.get("S3_SECURE", "true").lower() == "true" self._client: Any = None # lazy init def _get_client(self) -> Any: """Lazy-initialize the S3 client (minio or boto3).""" if self._client is not None: return self._client try: from minio import Minio # type: ignore self._client = Minio( endpoint=self.endpoint.replace("https://", "").replace("http://", ""), access_key=self.access_key, secret_key=self.secret_key, secure=self.secure, region=self.region, ) # Ensure bucket exists if not self._client.bucket_exists(self.bucket): self._client.make_bucket(self.bucket) logger.info("S3Storage: connected to %s, bucket=%s", self.endpoint, self.bucket) return self._client except ImportError: logger.error("S3Storage: minio package not installed. Install with: pip install minio") raise except Exception as e: logger.error("S3Storage: failed to connect to %s: %s", self.endpoint, e) raise async def save(self, path: str, data: bytes) -> str: from io import BytesIO client = self._get_client() client.put_object( bucket_name=self.bucket, object_name=path, data=BytesIO(data), length=len(data), ) logger.debug("S3Storage: saved %s (%d bytes)", path, len(data)) return path async def read(self, path: str) -> bytes: client = self._get_client() response = client.get_object(self.bucket, path) return response.read() async def delete(self, path: str) -> bool: client = self._get_client() try: client.remove_object(self.bucket, path) return True except Exception: return False async def exists(self, path: str) -> bool: client = self._get_client() try: client.stat_object(self.bucket, path) return True except Exception: return False async def get_url(self, path: str, expires: int = 3600) -> str: from datetime import timedelta client = self._get_client() return client.presigned_get_object(self.bucket, path, expires=timedelta(seconds=expires)) async def list_files(self, prefix: str) -> list[str]: client = self._get_client() objects = client.list_objects(self.bucket, prefix=prefix, recursive=True) return [obj.object_name for obj in objects] # ─── Factory ─── _storage_backend: StorageBackend | None = None def get_storage_backend() -> StorageBackend: """Get the configured storage backend singleton.""" global _storage_backend if _storage_backend is None: backend_type = os.environ.get("STORAGE_BACKEND", "local").lower() if backend_type == "s3": _storage_backend = S3Storage() logger.info("Storage backend: S3 (%s)", os.environ.get("S3_ENDPOINT", "")) else: _storage_backend = LocalStorage() logger.info("Storage backend: Local (%s)", os.environ.get("STORAGE_PATH", "/data/uploads")) return _storage_backend def reset_storage_backend() -> None: """Reset the storage backend singleton (for testing).""" global _storage_backend _storage_backend = None