Skip to content

anystore.tags

Tags

A simple tag implementation inspired by servicelayer.tags database

https://github.com/alephdata/servicelayer/blob/main/servicelayer/tags.py

It inherites the get and put methods from the base store and adds deletion by key filter criterion and a touch contextmanager.

Source code in anystore/tags.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class Tags:
    """
    A simple tag implementation inspired by `servicelayer.tags` database

    https://github.com/alephdata/servicelayer/blob/main/servicelayer/tags.py

    It inherites the `get` and `put` methods from the base store and adds
    deletion by key filter criterion and a `touch` contextmanager.
    """

    def __init__(self, store: BaseStore) -> None:
        self.store = store
        self.get = store.get
        self.put = store.put
        self.pop = store.pop
        self.exists = store.exists
        self.iterate_keys = store.iterate_keys
        self.iterate_values = store.iterate_values

    def delete(
        self,
        prefix: str | None = None,
        exclude_prefix: str | None = None,
        glob: str | None = None,
        ignore_errors: bool | None = False,
    ) -> None:
        """
        Delete values by given key criteria.

        Example:
            ```python
            tags.delete(prefix="runs/task1")
            ```

        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")
            ignore_errors: Ignore exceptions if deletion fails

        """
        for key in self.store.iterate_keys(
            prefix=prefix, exclude_prefix=exclude_prefix, glob=glob
        ):
            self.store.delete(key, ignore_errors)

    @contextlib.contextmanager
    def touch(self, key: Uri) -> Generator[datetime, None, None]:
        """
        Store the timestamp for a given tag but only at context leave.

        This is useful to tag a long running thing only after succeed but with a
        timestamp from when the action started.

        Example:
            ```python
            with tags.touch("run/1/succeed") as now:
                log.info(now)
                long_running_action_that_might_fail()
                # if it succeeds, the timestamp from the start (`now`) will be stored
            ```
        """
        now = datetime.now()
        try:
            yield now
        except Exception as e:
            raise e
        else:
            self.store.put(key, now)

delete(prefix=None, exclude_prefix=None, glob=None, ignore_errors=False)

Delete values by given key criteria.

Example 
tags.delete(prefix="runs/task1")

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
ignore_errors bool | None

Ignore exceptions if deletion fails

 False
Source code in anystore/tags.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def delete(
    self,
    prefix: str | None = None,
    exclude_prefix: str | None = None,
    glob: str | None = None,
    ignore_errors: bool | None = False,
) -> None:
    """
    Delete values by given key criteria.

    Example:
        ```python
        tags.delete(prefix="runs/task1")
        ```

    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")
        ignore_errors: Ignore exceptions if deletion fails

    """
    for key in self.store.iterate_keys(
        prefix=prefix, exclude_prefix=exclude_prefix, glob=glob
    ):
        self.store.delete(key, ignore_errors)

touch(key)

Store the timestamp for a given tag but only at context leave.

This is useful to tag a long running thing only after succeed but with a timestamp from when the action started.

Example 
with tags.touch("run/1/succeed") as now:
    log.info(now)
    long_running_action_that_might_fail()
    # if it succeeds, the timestamp from the start (`now`) will be stored
Source code in anystore/tags.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@contextlib.contextmanager
def touch(self, key: Uri) -> Generator[datetime, None, None]:
    """
    Store the timestamp for a given tag but only at context leave.

    This is useful to tag a long running thing only after succeed but with a
    timestamp from when the action started.

    Example:
        ```python
        with tags.touch("run/1/succeed") as now:
            log.info(now)
            long_running_action_that_might_fail()
            # if it succeeds, the timestamp from the start (`now`) will be stored
        ```
    """
    now = datetime.now()
    try:
        yield now
    except Exception as e:
        raise e
    else:
        self.store.put(key, now)