Bucket

amltk.store.bucket

Module containing the base definition of a bucket.

A bucket is a collection of resources that can be accessed by a key of a given type. This lets you easily store and retrieve objects of varying types in a single location.

Concrete examples

• PathBucket.

Bucket

Bases: ABC, MutableMapping[KeyT, Drop[LinkT]], Generic[KeyT, LinkT]

Definition of a bucket of resources, accessed by a Key.

Indexing into a bucket returns a Drop that can be used to access the resource.

The definition mostly follow that of MutableMapping, but with the change of .keys() and .values() to return iterators and .items() to return an iterator of tuples. The other change is that the .values() do not return the resources themselves, by rather a Drop which wraps the resource.

__contains__

__contains__(key: object) -> bool

Check if a key is in the bucket.

PARAMETER	DESCRIPTION
key	The key to check for. TYPE: object

Source code in src/amltk/store/bucket.py

@override
def __contains__(self, key: object) -> bool:
    """Check if a key is in the bucket.

    Args:
        key: The key to check for.
    """
    return key in self

__delitem__ abstractmethod

__delitem__(key: KeyT) -> None

Remove a resource from the bucket.

PARAMETER	DESCRIPTION
key	The key to the resource. TYPE: KeyT

Source code in src/amltk/store/bucket.py

@override
@abstractmethod
def __delitem__(self, key: KeyT) -> None:
    """Remove a resource from the bucket.

    Args:
        key: The key to the resource.
    """

__getitem__ abstractmethod

__getitem__(key: KeyT) -> Drop[LinkT]

Get a drop for a resource in the bucket.

PARAMETER	DESCRIPTION
key	The key to the resource. TYPE: KeyT

Source code in src/amltk/store/bucket.py

@override
@abstractmethod
def __getitem__(self, key: KeyT) -> Drop[LinkT]:
    """Get a drop for a resource in the bucket.

    Args:
        key: The key to the resource.
    """

__iter__ abstractmethod

__iter__() -> Iterator[KeyT]

Iterate over the keys in the bucket.

Source code in src/amltk/store/bucket.py

@override
@abstractmethod
def __iter__(self) -> Iterator[KeyT]:
    """Iterate over the keys in the bucket."""

__len__

__len__() -> int

Get the number of keys in the bucket.

Source code in src/amltk/store/bucket.py

@override
def __len__(self) -> int:
    """Get the number of keys in the bucket."""
    return ilen(iter(self))

__setitem__ abstractmethod

__setitem__(key: KeyT, value: Any) -> None

Store a value in the bucket.

PARAMETER	DESCRIPTION
key	The key to the resource. TYPE: KeyT
value	The value to store in the bucket. TYPE: Any

Source code in src/amltk/store/bucket.py

@override
@abstractmethod
def __setitem__(self, key: KeyT, value: Any) -> None:
    """Store a value in the bucket.

    Args:
        key: The key to the resource.
        value: The value to store in the bucket.
    """

fetch

fetch(
    *keys: KeyT,
    default: None | Any | dict[KeyT, Any] = None
) -> dict[KeyT, Any]

Fetch a resource from the bucket.

PARAMETER	DESCRIPTION
keys	The keys to the resources. TYPE: KeyT DEFAULT: ()
default	The default value to return if the key is not in the bucket. If a dict is passed, the default for each key will be the value in the dict for that key, using None if not present. TYPE: None | Any | dict[KeyT, Any] DEFAULT: None

RETURNS	DESCRIPTION
dict[KeyT, Any]	The resources stored in the bucket at the given keys.

Source code in src/amltk/store/bucket.py

def fetch(
    self,
    *keys: KeyT,
    default: None | Any | dict[KeyT, Any] = None,
) -> dict[KeyT, Any]:
    """Fetch a resource from the bucket.

    Args:
        keys: The keys to the resources.
        default: The default value to return if the key is not in the bucket.
            If a dict is passed, the default for each key will be the value
            in the dict for that key, using None if not present.

    Returns:
        The resources stored in the bucket at the given keys.
    """
    default_dict = {} if not isinstance(default, dict) else default
    return {
        key: self[key].get(default=default_dict.get(key, default)) for key in keys
    }

find

find(
    pattern: str, *, multi_key: bool = False
) -> (
    dict[str, Drop[LinkT]]
    | dict[tuple[str, ...], Drop[LinkT]]
    | None
)

Find resources in the bucket.

found = bucket.find(r"trial_(.+)_val_predictions.npy")  # (1)!
if found is None:
    raise KeyError("No predictions found")

for name, drop in found.items():
    predictions = drop.get()
    # Do something with the predictions
    # ...

1. The (.+) is a capture group which will attempt to match anything ., when there is one or more occurences +, and put it in a capure group (). What is captured will be used as the key in the returned dict.

PARAMETER	DESCRIPTION
pattern	The pattern to search for. TYPE: str
multi_key	Whether you have multiple capture groups in the pattern. Multiple capture groups with () If using multiple capture groups, the returned dict will have tuples as keys. If there is only one capture group, the tuple will be expanded to a single value. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
dict[str, Drop[LinkT]] | dict[tuple[str, ...], Drop[LinkT]] | None	A mapping of links to drops for the resources found.

Source code in src/amltk/store/bucket.py

def find(
    self,
    pattern: str,
    *,
    multi_key: bool = False,
) -> dict[str, Drop[LinkT]] | dict[tuple[str, ...], Drop[LinkT]] | None:
    """Find resources in the bucket.

    ```python
    found = bucket.find(r"trial_(.+)_val_predictions.npy")  # (1)!
    if found is None:
        raise KeyError("No predictions found")

    for name, drop in found.items():
        predictions = drop.get()
        # Do something with the predictions
        # ...
    ```

    1. The `(.+)` is a **capture group** which will attempt to match anything `.`,
        when there is one or more occurences `+`, and put it in a capure group `()`.
        What is captured will be used as the key in the returned dict.

    Args:
        pattern: The pattern to search for.
        multi_key: Whether you have multiple capture groups in the pattern.

            !!! note "Multiple capture groups with `()`"

                If using multiple capture groups, the returned dict will have
                tuples as keys. If there is only one capture group, the tuple
                will be expanded to a single value.

    Returns:
        A mapping of links to drops for the resources found.
    """
    keys = [(key, match) for key in self if (match := re.search(pattern, str(key)))]
    if not keys:
        return None

    matches = {match.groups(): self[key] for key, match in keys}

    # If it's a tuple of length 1, we expand it
    one_group = len(next(iter(matches.keys()))) == 1
    if one_group:
        if multi_key:
            raise ValueError(
                "Use multi_key=True when the pattern has more than 1 capture group",
            )

        return {key[0]: drop for key, drop in matches.items()}

    # Here we have multi-groups => tuple keys
    if not multi_key:
        raise ValueError(
            "Use multi_key=False when the pattern has only 1 capture group",
        )

    return matches

remove

remove(
    keys: Iterable[KeyT],
    *,
    how: Callable[[LinkT], bool] | None = None
) -> dict[KeyT, bool]

Remove resources from the bucket.

PARAMETER	DESCRIPTION
keys	The keys to the resources. TYPE: Iterable[KeyT]
how	A function that removes the resource. TYPE: Callable[[LinkT], bool] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[KeyT, bool]	A mapping of keys to whether they were removed.

Source code in src/amltk/store/bucket.py

def remove(
    self,
    keys: Iterable[KeyT],
    *,
    how: Callable[[LinkT], bool] | None = None,
) -> dict[KeyT, bool]:
    """Remove resources from the bucket.

    Args:
        keys: The keys to the resources.
        how: A function that removes the resource.

    Returns:
        A mapping of keys to whether they were removed.
    """
    return {key: self[key].remove(how=how) for key in keys}

store

store(other: Mapping[KeyT, Any]) -> None

Store items into the bucket with the given mapping.

PARAMETER	DESCRIPTION
other	The mapping of items to store in the bucket. TYPE: Mapping[KeyT, Any]

Source code in src/amltk/store/bucket.py

def store(self, other: Mapping[KeyT, Any]) -> None:
    """Store items into the bucket with the given mapping.

    Args:
        other: The mapping of items to store in the bucket.
    """
    for key, value in other.items():
        self[key] = value

sub abstractmethod

sub(key: KeyT) -> Self

Create a subbucket of this bucket.

PARAMETER	DESCRIPTION
key	The name of the sub bucket. TYPE: KeyT

RETURNS	DESCRIPTION
Self	A new bucket with the same loaders as the current bucket.

Source code in src/amltk/store/bucket.py

@abstractmethod
def sub(self, key: KeyT) -> Self:
    """Create a subbucket of this bucket.

    Args:
        key: The name of the sub bucket.

    Returns:
        A new bucket with the same loaders as the current bucket.
    """

update

update(items: Mapping[KeyT, Any]) -> None

Update the bucket with the given mapping.

PARAMETER	DESCRIPTION
items	The mapping of items to store in the bucket. TYPE: Mapping[KeyT, Any]

Source code in src/amltk/store/bucket.py

@override
def update(self, items: Mapping[KeyT, Any]) -> None:  # type: ignore
    """Update the bucket with the given mapping.

    Args:
        items: The mapping of items to store in the bucket.
    """
    for key, value in items.items():
        self[key].put(value)