feat: global storage configuration via file-keeper.json

Author: smotornyuk

pyproject.toml

@@ -39,6 +39,8 @@ test = [ "pytest-cov", "pytest-faker", "responses", "werkzeug", "faker" ]
 docs = [ "mkdocs", "mkdocs-material", "pymdown-extensions", "mkdocstrings[python]", ]
 dev = [ "pytest-cov", "pytest-faker", "responses", "mkdocs", "mkdocs-material", "pymdown-extensions", "mkdocstrings[python]",]
 
+user_config = ["platformdirs"]
+
 azure = ["azure-storage-blob",]
 gcs = [ "google-cloud-storage",]
 libcloud = [ "apache-libcloud", "cryptography",]

setup.py

@@ -1,4 +1,5 @@
 """Package definition."""
+
 from setuptools import setup
 
 setup()

src/file_keeper/__init__.py

@@ -1,4 +1,5 @@
 """Main entrypoint of the file-keeper."""
+
 __version__ = "0.1.0a0"
 
 from .core import exceptions as exc
@@ -12,6 +13,7 @@
     Storage,
     Uploader,
     adapters,
+    get_storage,
     make_storage,
 )
 from .core.types import Location, SignedAction
@@ -45,6 +47,7 @@
     "hookimpl",
     "humanize_filesize",
     "make_storage",
+    "get_storage",
     "make_upload",
     "parse_filesize",
     "types",

src/file_keeper/core/data.py

@@ -27,6 +27,7 @@
 @dataclasses.dataclass(frozen=True)
 class BaseData(Generic[TData]):
     """Base class for file details."""
+
     location: types.Location
     size: int = 0
     content_type: str = ""

src/file_keeper/core/registry.py

@@ -1,4 +1,5 @@
 """Registry for collections."""
+
 from __future__ import annotations
 
 from collections.abc import Callable, Hashable, Mapping, MutableMapping
@@ -78,6 +79,7 @@ def pop(self, key: K) -> V | None:
 
     def decorated(self, key: K):
         """Collect member via decorator."""
+
         def decorator(value: V):
             self.register(key, value)
             return value

src/file_keeper/core/storage.py

@@ -14,18 +14,32 @@
 import dataclasses
 import functools
 import inspect
+import json
 import logging
 import os
+import pathlib
 from abc import ABC
 from collections.abc import Callable, Iterable, Mapping
-from typing import Any, ClassVar, TypeAlias, cast
+from typing import Any, ClassVar, Literal, TypeAlias, cast
 
 from typing_extensions import ParamSpec, TypeVar, override
 
 from . import data, exceptions, types, utils
 from .registry import Registry
 from .upload import Upload, make_upload
 
+try:
+    from platformdirs import user_config_dir  # pyright: ignore[reportAssignmentType]
+except ImportError:
+
+    def user_config_dir(
+        appname: str | None = None,
+        appauthor: str | Literal[False] | None = None,
+    ):
+        """Mock for user config locator."""
+        return
+
+
 P = ParamSpec("P")
 T = TypeVar("T")
 S = TypeVar("S", bound="Storage")
@@ -36,6 +50,8 @@
 Capability: TypeAlias = utils.Capability
 
 adapters = Registry["type[Storage]"]()
+storages = Registry["Storage"]()
+
 location_transformers = Registry[types.LocationTransformer]()
 
 
@@ -995,3 +1011,42 @@ def make_storage(name: str, settings: dict[str, Any]) -> Storage:
     settings.setdefault("name", name)
 
     return adapter(settings)
+
+
+def get_storage(name: str, settings: dict[str, Any] | None = None) -> Storage:
+    """Get storage from the pool.
+
+    If storage accessed for the first time, it's initialized and added to the
+    pool. After that the same storage is returned every time the function is
+    called with the given name.
+
+    """
+    if name not in storages:
+        if settings is None:
+            config_file = os.getenv("FILE_KEEPER_CONFIG")
+
+            if not config_file and (config_dir := user_config_dir("file-keeper")):
+                config_file = os.path.join(config_dir, "file-keeper.json")
+
+            if not config_file or not os.path.isfile(config_file):
+                path = pathlib.Path().absolute()
+
+                while len(path.parts) > 1:
+                    config_file = str(path / "file-keeper.json")
+                    if os.path.exists(config_file):
+                        break
+                    path = path.parent
+                else:
+                    config_file = None
+
+            if config_file:
+                log.debug("Load configuration from %s", config_file)
+                with open(config_file) as src:
+                    settings = json.load(src).get("storages", {}).get(name)
+
+            if not settings:
+                raise exceptions.UnknownStorageError(name)
+
+        storages.register(name, make_storage(name, settings))
+
+    return storages[name]

src/file_keeper/core/types.py

@@ -1,4 +1,5 @@
 """Type definitions."""
+
 from __future__ import annotations
 
 from collections.abc import Callable, Iterator
@@ -17,16 +18,19 @@
 
 class PReadable(Protocol):
     """Readable object."""
+
     def read(self, size: Any = ..., /) -> bytes: ...
 
 
 class PStream(PReadable, Protocol):
     """Readable stream."""
+
     def __iter__(self) -> Iterator[bytes]: ...
 
 
 class PSeekableStream(PStream, Protocol):
     """Stream that supports `seek` operation."""
+
     def tell(self) -> int:
         """Get the current position of the pointer."""
         ...
@@ -38,6 +42,7 @@ def seek(self, offset: int, whence: int = 0) -> int:
 
 class PData(Protocol):
     """Structure of the *Data object."""
+
     location: Location
     size: int
     content_type: str

src/file_keeper/core/upload.py

@@ -1,4 +1,5 @@
 """Upload implementation."""
+
 from __future__ import annotations
 
 import dataclasses

src/file_keeper/core/utils.py

@@ -277,6 +277,7 @@ def humanize_filesize(value: int | float, base: int = SI_BASE) -> str:
 
 class AbstractReader(Generic[T], abc.ABC):
     """Abstract wrapper that transforms data into readable stream."""
+
     source: T
     chunk_size: int
 
@@ -296,6 +297,7 @@ def read(self, size: int | None = None) -> bytes:
 
 class IterableBytesReader(AbstractReader[Iterable[int]]):
     """Wrapper that transforms iterable of bytes into readable stream."""
+
     def __init__(self, source: Iterable[bytes], chunk_size: int = CHUNK_SIZE):
         super().__init__(itertools.chain.from_iterable(source), chunk_size)
 

src/file_keeper/default/__init__.py

@@ -1,4 +1,5 @@
 """Default implementations of file-keeper units."""
+
 from __future__ import annotations
 
 import contextlib