"""Storage backends and factory for notebook snapshots.""" from pathlib import Path from .base import StorageAdapter, validate_key from .local import LocalFileStorage from .s3 import S3ConnectionError, S3Storage def create_storage( storage_type: str, root_dir: str | Path, **kwargs, ) -> StorageAdapter: """Create a storage adapter from a configuration string. Parameters ---------- storage_type: str One of ``"local"`` or ``"s3"`` (covers AWS S3, MinIO, and any S3-API-compatible service via the ``endpoint`` parameter). root_dir: str | Path Root directory used by local storage. For S3 this is ignored. **kwargs Extra arguments forwarded to the backend constructor. For S3, expected keys include ``endpoint``, ``access_key``, ``secret_key``, ``bucket``, ``secure``, and ``prefix``. For local, ``local_root`` may override the default ``root_dir / ".notebook_versions"`` path. Returns ------- StorageAdapter Configured storage adapter instance. Raises ------ ValueError If ``storage_type`` is unknown. S3ConnectionError If ``storage_type="s3"`` and the endpoint is unreachable or credentials are missing/invalid. """ storage_type = storage_type.lower() if storage_type == "local": local_root = kwargs.get("local_root") if local_root: return LocalFileStorage(local_root) return LocalFileStorage(Path(root_dir) / ".notebook_versions") if storage_type == "s3": return S3Storage( endpoint=kwargs["endpoint"], access_key=kwargs["access_key"], secret_key=kwargs["secret_key"], bucket=kwargs.get("bucket", "notebook-snapshot"), secure=kwargs.get("secure", False), prefix=kwargs.get("prefix", ""), ) raise ValueError(f"Unknown storage type: {storage_type!r}") __all__ = [ "create_storage", "LocalFileStorage", "S3ConnectionError", "S3Storage", "StorageAdapter", "validate_key", ]