anystore.store

Top-level store entrypoint

get_store(uri=None, settings=None, **kwargs) cached

Short-hand initializer for a new store. The call is cached during runtime if input doesn't change.

Example

from anystore import get_store

# initialize from current configuration
store = get_store()
# get a redis store with custom prefix
store = get_store("redis://localhost", backend_config={"redis_prefix": "foo"})

Parameters:

Name	Type	Description	Default
uri	str | None	Store base uri, if relative it is considered as a local file store, otherwise the store backend is inferred from the scheme. If omitted, store is derived from settings defaults (taking current environment into account).	None
**kwargs	Any	pass through storage-specific options	{}

Returns:

Type	Description
BaseStore	A Store class

Source code in anystore/store/__init__.py

@cache
def get_store(
    uri: str | None = None, settings: Settings | None = None, **kwargs: Any
) -> BaseStore:
    """
    Short-hand initializer for a new store. The call is cached during runtime if
    input doesn't change.

    Example:
        ```python
        from anystore import get_store

        # initialize from current configuration
        store = get_store()
        # get a redis store with custom prefix
        store = get_store("redis://localhost", backend_config={"redis_prefix": "foo"})
        ```

    Args:
        uri: Store base uri, if relative it is considered as a local file store,
             otherwise the store backend is inferred from the scheme. If omitted,
             store is derived from settings defaults (taking current environment
             into account).
        **kwargs: pass through storage-specific options

    Returns:
        A `Store` class
    """
    settings = settings or Settings()
    if uri is None:
        if settings.yaml_uri is not None:
            store = BaseStore.from_yaml_uri(settings.yaml_uri, **kwargs)
            return get_store(**store.model_dump())
        if settings.json_uri is not None:
            store = BaseStore.from_json_uri(settings.json_uri, **kwargs)
            return get_store(**store.model_dump())
        uri = settings.uri
    uri = ensure_uri(uri)
    parsed = urlparse(uri)
    if parsed.scheme == "memory":
        return MemoryStore(uri=uri, **kwargs)
    if parsed.scheme == "redis":
        try:
            from anystore.store.redis import RedisStore

            return RedisStore(uri=uri, **kwargs)
        except ImportError as e:
            log.error("Install redis dependencies via `anystore[redis]`")
            raise ImportError(e)
    if "sql" in parsed.scheme:
        try:
            from anystore.store.sql import SqlStore

            return SqlStore(uri=uri, **kwargs)
        except ImportError as e:
            log.error("Install sql dependencies via `anystore[sql]`")
            raise ImportError(e)
    if "zip" in os.path.splitext(uri)[1]:
        return ZipStore(uri=uri, **kwargs)
    return Store(uri=uri, **kwargs)

Base store interface

The store class provides the top-level interface regardless for the storage backend.

BaseStore

Bases: StoreModel, AbstractBackend

Source code in anystore/store/base.py

class BaseStore(StoreModel, AbstractBackend):
    def get(
        self,
        key: Uri,
        raise_on_nonexist: bool | None = None,
        serialization_mode: Mode | None = None,
        deserialization_func: Callable | None = None,
        model: Model | None = None,
        **kwargs,
    ) -> Any:
        """
        Get a value from the store for the given key

        Args:
            key: Key relative to store base uri
            raise_on_nonexist: Raise `DoesNotExist` if key doesn't exist or stay
                silent, overrides store settings
            serialization_mode: Serialize result ("auto", "raw", "pickle",
                "json"), overrides store settings
            deserialization_func: Specific function to use (ignores
                `serialization_mode`), overrides store settings
            model: Pydantic serialization model (ignores `serialization_mode`
                and `deserialization_func`), overrides store settings

        Returns:
            The (optionally serialized) value for the key
        """
        serialization_mode = serialization_mode or self.serialization_mode
        deserialization_func = deserialization_func or self.deserialization_func
        model = model or self.model
        if raise_on_nonexist is None:
            raise_on_nonexist = self.raise_on_nonexist
        kwargs = self.ensure_kwargs(**kwargs)
        key = self.get_key(key)
        try:
            return from_store(
                self._read(key, raise_on_nonexist, **kwargs),
                serialization_mode,
                deserialization_func=deserialization_func,
                model=model,
            )
        except FileNotFoundError:  # fsspec
            if raise_on_nonexist:
                raise DoesNotExist(f"Key does not exist: `{key}`")
            return None

    @check_readonly
    def pop(self, key: Uri, *args: Any, **kwargs: Any) -> Any:
        """
        Retrieve the value for the given key and remove it from the store.

        Args:
            key: Key relative to store base uri
            *args: Any valid arguments for the stores `get` function
            **kwargs: Any valid arguments for the stores `get` function

        Returns:
            The (optionally serialized) value for the key
        """
        value = self.get(key, *args, **kwargs)
        self._delete(self.get_key(key))
        return value

    @check_readonly
    def delete(self, key: Uri, ignore_errors: bool = False) -> None:
        """
        Delete the content at the given key.

        Args:
            key: Key relative to store base uri
            ignore_errors: Ignore exceptions if deletion fails
        """
        try:
            self._delete(self.get_key(key))
        except Exception as e:
            if not ignore_errors:
                raise e

    def stream(
        self,
        key: Uri,
        raise_on_nonexist: bool | None = None,
        serialization_mode: Mode | None = None,
        deserialization_func: Callable | None = None,
        model: Model | None = None,
        **kwargs,
    ) -> Generator[Any, None, None]:
        """
        Stream a value line by line from the store for the given key

        Args:
            key: Key relative to store base uri
            raise_on_nonexist: Raise `DoesNotExist` if key doesn't exist or stay
                silent, overrides store settings
            serialization_mode: Serialize result ("auto", "raw", "pickle",
                "json"), overrides store settings
            deserialization_func: Specific function to use (ignores
                `serialization_mode`), overrides store settings
            model: Pydantic serialization model (ignores `serialization_mode`
                and `deserialization_func`), overrides store settings

        Yields:
            The (optionally serialized) values line by line

        Raises:
            anystore.exceptions.DoesNotExists: If key doesn't exist and
                raise_on_nonexist=True
        """
        key = self.get_key(key)
        model = model or self.model
        extra_kwargs = {
            "serialization_mode": serialization_mode or self.serialization_mode,
            "deserialization_func": deserialization_func or self.deserialization_func,
            "model": model,
        }
        try:
            for line in self._stream(key, **kwargs):
                yield from_store(line, **extra_kwargs)
        except (FileNotFoundError, DoesNotExist):
            if raise_on_nonexist:
                raise DoesNotExist(f"Key does not exist: `{key}`")
            return None

    @check_readonly
    def put(
        self,
        key: Uri,
        value: Any,
        serialization_mode: Mode | None = None,
        serialization_func: Callable | None = None,
        model: Model | None = None,
        ttl: int | None = None,
        **kwargs,
    ):
        """
        Store a value at the given key

        Args:
            key: Key relative to store base uri
            value: The content
            serialization_mode: Serialize value prior to storing ("auto", "raw",
                "pickle", "json"), overrides store settings
            serialization_func: Specific function to use (ignores
                `serialization_mode`), overrides store settings
            model: Pydantic serialization model (ignores `serialization_mode`
                and `deserialization_func`), overrides store settings
            ttl: Time to live (in seconds) for that key if the backend supports
                it (e.g. redis, sql)
        """
        serialization_mode = serialization_mode or self.serialization_mode
        serialization_func = serialization_func or self.serialization_func
        model = model or self.model
        kwargs = self.ensure_kwargs(**kwargs)
        key = self.get_key(key)
        ttl = ttl or self.default_ttl or None
        self._write(
            key,
            to_store(
                value,
                serialization_mode,
                serialization_func=serialization_func,
                model=model,
            ),
            ttl=ttl,
        )

    def exists(self, key: Uri) -> bool:
        """Check if the given `key` exists"""
        return self._exists(self.get_key(key))

    def info(self, key: Uri) -> Stats:
        """
        Get metadata for the given `key`.

        Returns:
            Key metadata
        """
        stats = self._info(self.get_key(key))
        key = str(key)
        return Stats(
            **stats.model_dump(),
            name=Path(key).name,
            store=str(self.uri),
            key=key,
        )

    def ensure_kwargs(self, **kwargs) -> dict[str, Any]:
        config = clean_dict(self.backend_config)
        return {**config, **clean_dict(kwargs)}

    def get_key(self, key: Uri) -> str:
        return unquote(f"{self._get_key_prefix()}/{str(key)}".strip("/"))

    def iterate_keys(
        self,
        prefix: str | None = None,
        exclude_prefix: str | None = None,
        glob: str | None = None,
    ) -> Generator[str, None, None]:
        """
        Iterate through all the keys in the store based on given criteria.
        Criteria can be combined (e.g. include but exclude a subset).

        Example:
            ```python
            for key in store.iterate_keys(prefix="dataset1", glob="*.pdf"):
                data = store.get(key, mode="raw")
                parse(data)
            ```

        Args:
            prefix: Include only keys with the given prefix (e.g. "foo/bar")
            exclude_prefix: Exclude keys with this prefix
            glob: Path-style glob pattern for keys to filter (e.g. "foo/**/*.json")

        Returns:
            The matching keys as a generator of strings
        """
        for key in self._iterate_keys(prefix, exclude_prefix, glob):
            yield unquote(key)

    def checksum(
        self, key: Uri, algorithm: str | None = DEFAULT_HASH_ALGORITHM, **kwargs: Any
    ) -> str:
        """
        Get the checksum for the value at the given key

        Args:
            key: Key relative to store base uri
            algorithm: Checksum algorithm from `hashlib` (default: "sha1")
            **kwargs: Pass through arguments to content retrieval

        Returns:
            The computed checksum
        """
        kwargs = self.ensure_kwargs(**kwargs)
        kwargs["mode"] = "rb"
        key = self.get_key(key)
        with self._open(key, **kwargs) as io:
            return make_checksum(io, algorithm or DEFAULT_HASH_ALGORITHM)

    def open(
        self, key: Uri, mode: str | None = DEFAULT_MODE, **kwargs: Any
    ) -> Generator[IO, None, None]:
        """
        Open the given key similar to built-in `open()`

        Example:
            ```python
            from anystore import get_store

            store = get_store()
            with store.open("foo/bar.txt") as fh:
                return fh.read()
            ```

        Args:
            key: Key relative to store base uri
            mode: Open mode ("rb", "wb", "r", "w")
            **kwargs: Pass through arguments to backend

        Returns:
            The open handler
        """
        mode = mode or DEFAULT_MODE
        if self.readonly and ("w" in mode or "a" in mode):
            raise ReadOnlyError(f"Store `{self.uri}` is configured readonly!")
        kwargs = self.ensure_kwargs(**kwargs)
        key = self.get_key(key)
        return self._open(key, mode=mode, **kwargs)

    @check_readonly
    def touch(self, key: Uri, **kwargs: Any) -> datetime:
        """
        Store the current timestamp at the given key

        Args:
            key: Key relative to store base uri
            **kwargs: Any valid arguments for the stores `put` function

        Returns:
            The timestamp
        """
        now = datetime.now()
        self.put(key, now, **kwargs)
        return now

checksum(key, algorithm=DEFAULT_HASH_ALGORITHM, **kwargs)

Get the checksum for the value at the given key

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
algorithm	str | None	Checksum algorithm from hashlib (default: "sha1")	DEFAULT_HASH_ALGORITHM
**kwargs	Any	Pass through arguments to content retrieval	{}

Returns:

Type	Description
str	The computed checksum

Source code in anystore/store/base.py

def checksum(
    self, key: Uri, algorithm: str | None = DEFAULT_HASH_ALGORITHM, **kwargs: Any
) -> str:
    """
    Get the checksum for the value at the given key

    Args:
        key: Key relative to store base uri
        algorithm: Checksum algorithm from `hashlib` (default: "sha1")
        **kwargs: Pass through arguments to content retrieval

    Returns:
        The computed checksum
    """
    kwargs = self.ensure_kwargs(**kwargs)
    kwargs["mode"] = "rb"
    key = self.get_key(key)
    with self._open(key, **kwargs) as io:
        return make_checksum(io, algorithm or DEFAULT_HASH_ALGORITHM)

delete(key, ignore_errors=False)

Delete the content at the given key.

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
ignore_errors	bool	Ignore exceptions if deletion fails	False

Source code in anystore/store/base.py

@check_readonly
def delete(self, key: Uri, ignore_errors: bool = False) -> None:
    """
    Delete the content at the given key.

    Args:
        key: Key relative to store base uri
        ignore_errors: Ignore exceptions if deletion fails
    """
    try:
        self._delete(self.get_key(key))
    except Exception as e:
        if not ignore_errors:
            raise e

exists(key)

Check if the given key exists

Source code in anystore/store/base.py

def exists(self, key: Uri) -> bool:
    """Check if the given `key` exists"""
    return self._exists(self.get_key(key))

get(key, raise_on_nonexist=None, serialization_mode=None, deserialization_func=None, model=None, **kwargs)

Get a value from the store for the given key

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
raise_on_nonexist	bool | None	Raise DoesNotExist if key doesn't exist or stay silent, overrides store settings	None
serialization_mode	Mode | None	Serialize result ("auto", "raw", "pickle", "json"), overrides store settings	None
deserialization_func	Callable | None	Specific function to use (ignores serialization_mode), overrides store settings	None
model	Model | None	Pydantic serialization model (ignores serialization_mode and deserialization_func), overrides store settings	None

Returns:

Type	Description
Any	The (optionally serialized) value for the key

Source code in anystore/store/base.py

def get(
    self,
    key: Uri,
    raise_on_nonexist: bool | None = None,
    serialization_mode: Mode | None = None,
    deserialization_func: Callable | None = None,
    model: Model | None = None,
    **kwargs,
) -> Any:
    """
    Get a value from the store for the given key

    Args:
        key: Key relative to store base uri
        raise_on_nonexist: Raise `DoesNotExist` if key doesn't exist or stay
            silent, overrides store settings
        serialization_mode: Serialize result ("auto", "raw", "pickle",
            "json"), overrides store settings
        deserialization_func: Specific function to use (ignores
            `serialization_mode`), overrides store settings
        model: Pydantic serialization model (ignores `serialization_mode`
            and `deserialization_func`), overrides store settings

    Returns:
        The (optionally serialized) value for the key
    """
    serialization_mode = serialization_mode or self.serialization_mode
    deserialization_func = deserialization_func or self.deserialization_func
    model = model or self.model
    if raise_on_nonexist is None:
        raise_on_nonexist = self.raise_on_nonexist
    kwargs = self.ensure_kwargs(**kwargs)
    key = self.get_key(key)
    try:
        return from_store(
            self._read(key, raise_on_nonexist, **kwargs),
            serialization_mode,
            deserialization_func=deserialization_func,
            model=model,
        )
    except FileNotFoundError:  # fsspec
        if raise_on_nonexist:
            raise DoesNotExist(f"Key does not exist: `{key}`")
        return None

info(key)

Get metadata for the given key.

Returns:

Type	Description
Stats	Key metadata

Source code in anystore/store/base.py

def info(self, key: Uri) -> Stats:
    """
    Get metadata for the given `key`.

    Returns:
        Key metadata
    """
    stats = self._info(self.get_key(key))
    key = str(key)
    return Stats(
        **stats.model_dump(),
        name=Path(key).name,
        store=str(self.uri),
        key=key,
    )

iterate_keys(prefix=None, exclude_prefix=None, glob=None)

Iterate through all the keys in the store based on given criteria. Criteria can be combined (e.g. include but exclude a subset).

Example

for key in store.iterate_keys(prefix="dataset1", glob="*.pdf"):
    data = store.get(key, mode="raw")
    parse(data)

Parameters:

Name	Type	Description	Default
prefix	str | None	Include only keys with the given prefix (e.g. "foo/bar")	None
exclude_prefix	str | None	Exclude keys with this prefix	None
glob	str | None	Path-style glob pattern for keys to filter (e.g. "foo/*/.json")	None

Returns:

Type	Description
None	The matching keys as a generator of strings

Source code in anystore/store/base.py

def iterate_keys(
    self,
    prefix: str | None = None,
    exclude_prefix: str | None = None,
    glob: str | None = None,
) -> Generator[str, None, None]:
    """
    Iterate through all the keys in the store based on given criteria.
    Criteria can be combined (e.g. include but exclude a subset).

    Example:
        ```python
        for key in store.iterate_keys(prefix="dataset1", glob="*.pdf"):
            data = store.get(key, mode="raw")
            parse(data)
        ```

    Args:
        prefix: Include only keys with the given prefix (e.g. "foo/bar")
        exclude_prefix: Exclude keys with this prefix
        glob: Path-style glob pattern for keys to filter (e.g. "foo/**/*.json")

    Returns:
        The matching keys as a generator of strings
    """
    for key in self._iterate_keys(prefix, exclude_prefix, glob):
        yield unquote(key)

open(key, mode=DEFAULT_MODE, **kwargs)

Open the given key similar to built-in open()

Example

from anystore import get_store

store = get_store()
with store.open("foo/bar.txt") as fh:
    return fh.read()

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
mode	str | None	Open mode ("rb", "wb", "r", "w")	DEFAULT_MODE
**kwargs	Any	Pass through arguments to backend	{}

Returns:

Type	Description
None	The open handler

Source code in anystore/store/base.py

def open(
    self, key: Uri, mode: str | None = DEFAULT_MODE, **kwargs: Any
) -> Generator[IO, None, None]:
    """
    Open the given key similar to built-in `open()`

    Example:
        ```python
        from anystore import get_store

        store = get_store()
        with store.open("foo/bar.txt") as fh:
            return fh.read()
        ```

    Args:
        key: Key relative to store base uri
        mode: Open mode ("rb", "wb", "r", "w")
        **kwargs: Pass through arguments to backend

    Returns:
        The open handler
    """
    mode = mode or DEFAULT_MODE
    if self.readonly and ("w" in mode or "a" in mode):
        raise ReadOnlyError(f"Store `{self.uri}` is configured readonly!")
    kwargs = self.ensure_kwargs(**kwargs)
    key = self.get_key(key)
    return self._open(key, mode=mode, **kwargs)

pop(key, *args, **kwargs)

Retrieve the value for the given key and remove it from the store.

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
*args	Any	Any valid arguments for the stores get function	()
**kwargs	Any	Any valid arguments for the stores get function	{}

Returns:

Type	Description
Any	The (optionally serialized) value for the key

Source code in anystore/store/base.py

@check_readonly
def pop(self, key: Uri, *args: Any, **kwargs: Any) -> Any:
    """
    Retrieve the value for the given key and remove it from the store.

    Args:
        key: Key relative to store base uri
        *args: Any valid arguments for the stores `get` function
        **kwargs: Any valid arguments for the stores `get` function

    Returns:
        The (optionally serialized) value for the key
    """
    value = self.get(key, *args, **kwargs)
    self._delete(self.get_key(key))
    return value

put(key, value, serialization_mode=None, serialization_func=None, model=None, ttl=None, **kwargs)

Store a value at the given key

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
value	Any	The content	required
serialization_mode	Mode | None	Serialize value prior to storing ("auto", "raw", "pickle", "json"), overrides store settings	None
serialization_func	Callable | None	Specific function to use (ignores serialization_mode), overrides store settings	None
model	Model | None	Pydantic serialization model (ignores serialization_mode and deserialization_func), overrides store settings	None
ttl	int | None	Time to live (in seconds) for that key if the backend supports it (e.g. redis, sql)	None

Source code in anystore/store/base.py

@check_readonly
def put(
    self,
    key: Uri,
    value: Any,
    serialization_mode: Mode | None = None,
    serialization_func: Callable | None = None,
    model: Model | None = None,
    ttl: int | None = None,
    **kwargs,
):
    """
    Store a value at the given key

    Args:
        key: Key relative to store base uri
        value: The content
        serialization_mode: Serialize value prior to storing ("auto", "raw",
            "pickle", "json"), overrides store settings
        serialization_func: Specific function to use (ignores
            `serialization_mode`), overrides store settings
        model: Pydantic serialization model (ignores `serialization_mode`
            and `deserialization_func`), overrides store settings
        ttl: Time to live (in seconds) for that key if the backend supports
            it (e.g. redis, sql)
    """
    serialization_mode = serialization_mode or self.serialization_mode
    serialization_func = serialization_func or self.serialization_func
    model = model or self.model
    kwargs = self.ensure_kwargs(**kwargs)
    key = self.get_key(key)
    ttl = ttl or self.default_ttl or None
    self._write(
        key,
        to_store(
            value,
            serialization_mode,
            serialization_func=serialization_func,
            model=model,
        ),
        ttl=ttl,
    )

stream(key, raise_on_nonexist=None, serialization_mode=None, deserialization_func=None, model=None, **kwargs)

Stream a value line by line from the store for the given key

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
raise_on_nonexist	bool | None	Raise DoesNotExist if key doesn't exist or stay silent, overrides store settings	None
serialization_mode	Mode | None	Serialize result ("auto", "raw", "pickle", "json"), overrides store settings	None
deserialization_func	Callable | None	Specific function to use (ignores serialization_mode), overrides store settings	None
model	Model | None	Pydantic serialization model (ignores serialization_mode and deserialization_func), overrides store settings	None

Yields:

Type	Description
Any	The (optionally serialized) values line by line

Raises:

Type	Description
DoesNotExists	If key doesn't exist and raise_on_nonexist=True

Source code in anystore/store/base.py

def stream(
    self,
    key: Uri,
    raise_on_nonexist: bool | None = None,
    serialization_mode: Mode | None = None,
    deserialization_func: Callable | None = None,
    model: Model | None = None,
    **kwargs,
) -> Generator[Any, None, None]:
    """
    Stream a value line by line from the store for the given key

    Args:
        key: Key relative to store base uri
        raise_on_nonexist: Raise `DoesNotExist` if key doesn't exist or stay
            silent, overrides store settings
        serialization_mode: Serialize result ("auto", "raw", "pickle",
            "json"), overrides store settings
        deserialization_func: Specific function to use (ignores
            `serialization_mode`), overrides store settings
        model: Pydantic serialization model (ignores `serialization_mode`
            and `deserialization_func`), overrides store settings

    Yields:
        The (optionally serialized) values line by line

    Raises:
        anystore.exceptions.DoesNotExists: If key doesn't exist and
            raise_on_nonexist=True
    """
    key = self.get_key(key)
    model = model or self.model
    extra_kwargs = {
        "serialization_mode": serialization_mode or self.serialization_mode,
        "deserialization_func": deserialization_func or self.deserialization_func,
        "model": model,
    }
    try:
        for line in self._stream(key, **kwargs):
            yield from_store(line, **extra_kwargs)
    except (FileNotFoundError, DoesNotExist):
        if raise_on_nonexist:
            raise DoesNotExist(f"Key does not exist: `{key}`")
        return None

touch(key, **kwargs)

Store the current timestamp at the given key

Parameters:

Name	Type	Description	Default
key	Uri	Key relative to store base uri	required
**kwargs	Any	Any valid arguments for the stores put function	{}

Returns:

Type	Description
datetime	The timestamp

Source code in anystore/store/base.py

@check_readonly
def touch(self, key: Uri, **kwargs: Any) -> datetime:
    """
    Store the current timestamp at the given key

    Args:
        key: Key relative to store base uri
        **kwargs: Any valid arguments for the stores `put` function

    Returns:
        The timestamp
    """
    now = datetime.now()
    self.put(key, now, **kwargs)
    return now

VirtualIOMixin

Fake open() method for non file-like backends

Source code in anystore/store/base.py

class VirtualIOMixin:
    """
    Fake `open()` method for non file-like backends
    """

    @contextlib.contextmanager
    def _open(self, key: str, **kwargs) -> Generator[BytesIO | StringIO, None, None]:
        mode = kwargs.get("mode", DEFAULT_MODE)
        writer = "w" in mode
        if not writer:
            content = self._read(key, **kwargs)
            if "b" in mode:
                handler = BytesIO(content)
            else:
                handler = StringIO(content)
        else:
            if "b" in mode:
                handler = BytesIO()
            else:
                handler = StringIO()
        try:
            yield handler
        finally:
            if writer:
                self._write(key, handler.getvalue(), **kwargs)
            handler.close()

check_readonly(func)

Guard for read-only store. Write functions should be decorated with it

Source code in anystore/store/base.py

def check_readonly(func: Callable):
    """Guard for read-only store. Write functions should be decorated with it"""

    def _check(store: "BaseStore", *args, **kwargs):
        if store.readonly:
            raise ReadOnlyError(f"Store `{store.uri}` is configured readonly!")
        return func(store, *args, **kwargs)

    return _check

Models

Pydantic model interfaces to initialize stores and handle metadata for keys.

BaseStats

Bases: BaseModel

Shared base metadata object

Source code in anystore/model.py

class BaseStats(BaseModel):
    """Shared base metadata object"""

    created_at: datetime | None = None
    """Created at timestamp"""

    updated_at: datetime | None = None
    """Last updated timestamp"""

    size: int
    """Size (content length) in bytes"""

created_at: datetime | None = None class-attribute instance-attribute

Created at timestamp

size: int instance-attribute

Size (content length) in bytes

updated_at: datetime | None = None class-attribute instance-attribute

Last updated timestamp

Stats

Bases: BaseStats

Meta information for a store key

Source code in anystore/model.py

class Stats(BaseStats):
    """Meta information for a store key"""

    name: str
    """Key name: last part of the key (aka file name without path)"""
    store: str
    """Store base uri"""
    key: str
    """Full path of key"""

    @property
    def uri(self) -> str:
        """
        Computed uri property. Absolute when file-like prepended with store
        schema, relative if using different store backend

        Returns:
            file-like: `file:///tmp/foo.txt`, `ssh://user@host:data.csv`
            relative path for other (redis, sql, ...): `tmp/foo.txt`
        """
        store = StoreModel(uri=self.store)
        if store.is_fslike:
            return join_uri(self.store, self.key)
        return self.key

key: str instance-attribute

Full path of key

name: str instance-attribute

Key name: last part of the key (aka file name without path)

store: str instance-attribute

Store base uri

uri: str property

Computed uri property. Absolute when file-like prepended with store schema, relative if using different store backend

Returns:

Type	Description
str	file-like: file:///tmp/foo.txt, ssh://user@host:data.csv
str	relative path for other (redis, sql, ...): tmp/foo.txt

StoreModel

Bases: BaseModel

Store model to initialize a store from configuration

Source code in anystore/model.py

class StoreModel(BaseModel):
    """Store model to initialize a store from configuration"""

    uri: str
    """Store base uri"""
    serialization_mode: Mode | None = settings.serialization_mode
    """Default serialization (auto, raw, pickle, json)"""
    serialization_func: Callable | None = None
    """Default serialization function"""
    deserialization_func: Callable | None = None
    """Default deserialization function"""
    model: Model | None = None
    """Default pydantic model for serialization"""
    raise_on_nonexist: bool | None = settings.raise_on_nonexist
    """Raise `anystore.exceptions.DoesNotExist` if key doesn't exist"""
    default_ttl: int | None = settings.default_ttl
    """Default ttl for keys (only backends that support it: redis, sql, ..)"""
    backend_config: dict[str, Any] | None = None
    """Backend-specific configuration to pass through for initialization"""
    readonly: bool | None = False
    """Consider this store as a read-only store, writing will raise an exception"""

    @cached_property
    def scheme(self) -> str:
        return urlparse(self.uri).scheme

    @cached_property
    def path(self) -> str:
        return urlparse(self.uri).path.strip("/")

    @cached_property
    def netloc(self) -> str:
        return urlparse(self.uri).netloc

    @cached_property
    def is_local(self) -> bool:
        """Check if it is a local file store"""
        return self.scheme == SCHEME_FILE

    @cached_property
    def is_fslike(self) -> bool:
        """Check if it is a file-like store usable with `fsspec`"""
        return not self.is_sql and self.scheme not in (SCHEME_REDIS, SCHEME_MEMORY)

    @cached_property
    def is_sql(self) -> bool:
        """Check if it is a sql-like store (sqlite, postgres, ...)"""
        return "sql" in self.scheme

    @field_validator("uri", mode="before")
    @classmethod
    def ensure_uri(cls, v: Any) -> str:
        uri = ensure_uri(v)
        return uri.rstrip("/")

backend_config: dict[str, Any] | None = None class-attribute instance-attribute

Backend-specific configuration to pass through for initialization

default_ttl: int | None = settings.default_ttl class-attribute instance-attribute

Default ttl for keys (only backends that support it: redis, sql, ..)

deserialization_func: Callable | None = None class-attribute instance-attribute

Default deserialization function

is_fslike: bool cached property

Check if it is a file-like store usable with fsspec

is_local: bool cached property

Check if it is a local file store

is_sql: bool cached property

Check if it is a sql-like store (sqlite, postgres, ...)

model: Model | None = None class-attribute instance-attribute

Default pydantic model for serialization

raise_on_nonexist: bool | None = settings.raise_on_nonexist class-attribute instance-attribute

Raise anystore.exceptions.DoesNotExist if key doesn't exist

readonly: bool | None = False class-attribute instance-attribute

Consider this store as a read-only store, writing will raise an exception

serialization_func: Callable | None = None class-attribute instance-attribute

Default serialization function

serialization_mode: Mode | None = settings.serialization_mode class-attribute instance-attribute

Default serialization (auto, raw, pickle, json)

uri: str instance-attribute

Store base uri