Skip to content

Core Caching

Cache dataclass

Bases: Generic[EndpointDefinitionGen, TResponse]

Source code in src/pomdapi/core/caching.py 
 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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@dataclass
class Cache(Generic[EndpointDefinitionGen, TResponse]):
    _backend: CacheBackend
    _ttl: int = 60

    @staticmethod
    def key_from_req(endpoint_name: str, request: EndpointDefinitionGen) -> str:
        return f"{endpoint_name}/{request}"

    @staticmethod
    def key_from_tag(tag: str | Tag) -> str:
        return f"tag/{tag}"

    def get_by_request(
        self,
        endpoint_name: str,
        request: EndpointDefinitionGen,
    ) -> Optional[TResponse]:
        key = self.key_from_req(endpoint_name, request)
        return self._backend.get(key)

    async def aget_by_request(
        self,
        endpoint_name: str,
        request: EndpointDefinitionGen,
    ) -> Optional[TResponse]:
        """Get a response from the cache by request."""
        key = self.key_from_req(endpoint_name, request)
        return await self._backend.aget(key)

    def get_by_tags(
        self,
        endpoint_name: str,
        tags: Iterable[str | Tag],
    ) -> Optional[TResponse]:
        """Get a response from the cache by tags."""
        for tag in tags:
            key = self.key_from_tag(tag)
            if response := self._backend.get(key):
                return response

    async def aget_by_tags(
        self,
        endpoint_name: str,
        tags: Iterable[str | Tag],
    ) -> Optional[TResponse]:
        """Get a response from the cache by tags."""
        for tag in tags:
            key = self.key_from_tag(tag)
            if response := await self._backend.aget(key):
                return response

    def set(
        self,
        endpoint_name: str,
        request: EndpointDefinitionGen,
        tags: Iterable[str | Tag],
        response: TResponse,
        ttl: Optional[int] = None,
    ) -> None:
        """Set a response in the cache."""
        request_key = self.key_from_req(endpoint_name, request)
        self._backend.set(request_key, response, ttl=ttl)
        for tag in tags:
            key = self.key_from_tag(tag)
            self._backend.set(key, request_key, ttl=ttl)

    async def aset(
        self,
        endpoint_name: str,
        request: EndpointDefinitionGen,
        tags: Iterable[str | Tag],
        response: TResponse,
        ttl: Optional[int] = None,
    ) -> None:
        """Set a response in the cache."""
        request_def_key = self.key_from_req(endpoint_name, request)
        async with asyncio.TaskGroup() as tg:
            tg.create_task(self._backend.aset(request_def_key, response, ttl=ttl))
            for tag in tags:
                key = self.key_from_tag(tag)
                tg.create_task(self._backend.aset(key, request_def_key, ttl=ttl))

    def invalidate_tags(self, endpoint_name: str, tags: Iterable[str | Tag]) -> None:
        """Invalidate a response from the cache by tags."""
        for i, tag in enumerate(tags):
            tag_key = self.key_from_tag(tag)
            if i == 0:
                if (request_key := self._backend.get(tag_key)):
                    self._backend.delete(request_key)
            self._backend.delete(tag_key)

    async def ainvalidate_tags(
        self, endpoint_name: str, tags: Iterable[str | Tag]
    ) -> None:
        """Invalidate a response from the cache by tags."""
        async with asyncio.TaskGroup() as tg:
            for i, tag in enumerate(tags):
                tag_key = self.key_from_tag(tag)
                if i == 0:
                    if (request_key := await self._backend.aget(tag_key)):
                        tg.create_task(self._backend.adelete(request_key))
                tg.create_task(self._backend.adelete(tag_key))

aget_by_request(endpoint_name, request) async

Get a response from the cache by request.

Source code in src/pomdapi/core/caching.py 
70
71
72
73
74
75
76
77
async def aget_by_request(
    self,
    endpoint_name: str,
    request: EndpointDefinitionGen,
) -> Optional[TResponse]:
    """Get a response from the cache by request."""
    key = self.key_from_req(endpoint_name, request)
    return await self._backend.aget(key)

aget_by_tags(endpoint_name, tags) async

Get a response from the cache by tags.

Source code in src/pomdapi/core/caching.py 
90
91
92
93
94
95
96
97
98
99
async def aget_by_tags(
    self,
    endpoint_name: str,
    tags: Iterable[str | Tag],
) -> Optional[TResponse]:
    """Get a response from the cache by tags."""
    for tag in tags:
        key = self.key_from_tag(tag)
        if response := await self._backend.aget(key):
            return response

ainvalidate_tags(endpoint_name, tags) async

Invalidate a response from the cache by tags.

Source code in src/pomdapi/core/caching.py 
141
142
143
144
145
146
147
148
149
150
151
async def ainvalidate_tags(
    self, endpoint_name: str, tags: Iterable[str | Tag]
) -> None:
    """Invalidate a response from the cache by tags."""
    async with asyncio.TaskGroup() as tg:
        for i, tag in enumerate(tags):
            tag_key = self.key_from_tag(tag)
            if i == 0:
                if (request_key := await self._backend.aget(tag_key)):
                    tg.create_task(self._backend.adelete(request_key))
            tg.create_task(self._backend.adelete(tag_key))

aset(endpoint_name, request, tags, response, ttl=None) async

Set a response in the cache.

Source code in src/pomdapi/core/caching.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def aset(
    self,
    endpoint_name: str,
    request: EndpointDefinitionGen,
    tags: Iterable[str | Tag],
    response: TResponse,
    ttl: Optional[int] = None,
) -> None:
    """Set a response in the cache."""
    request_def_key = self.key_from_req(endpoint_name, request)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(self._backend.aset(request_def_key, response, ttl=ttl))
        for tag in tags:
            key = self.key_from_tag(tag)
            tg.create_task(self._backend.aset(key, request_def_key, ttl=ttl))

get_by_tags(endpoint_name, tags)

Get a response from the cache by tags.

Source code in src/pomdapi/core/caching.py 
79
80
81
82
83
84
85
86
87
88
def get_by_tags(
    self,
    endpoint_name: str,
    tags: Iterable[str | Tag],
) -> Optional[TResponse]:
    """Get a response from the cache by tags."""
    for tag in tags:
        key = self.key_from_tag(tag)
        if response := self._backend.get(key):
            return response

invalidate_tags(endpoint_name, tags)

Invalidate a response from the cache by tags.

Source code in src/pomdapi/core/caching.py 
132
133
134
135
136
137
138
139
def invalidate_tags(self, endpoint_name: str, tags: Iterable[str | Tag]) -> None:
    """Invalidate a response from the cache by tags."""
    for i, tag in enumerate(tags):
        tag_key = self.key_from_tag(tag)
        if i == 0:
            if (request_key := self._backend.get(tag_key)):
                self._backend.delete(request_key)
        self._backend.delete(tag_key)

set(endpoint_name, request, tags, response, ttl=None)

Set a response in the cache.

Source code in src/pomdapi/core/caching.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def set(
    self,
    endpoint_name: str,
    request: EndpointDefinitionGen,
    tags: Iterable[str | Tag],
    response: TResponse,
    ttl: Optional[int] = None,
) -> None:
    """Set a response in the cache."""
    request_key = self.key_from_req(endpoint_name, request)
    self._backend.set(request_key, response, ttl=ttl)
    for tag in tags:
        key = self.key_from_tag(tag)
        self._backend.set(key, request_key, ttl=ttl)

CacheBackend

Bases: Protocol

Protocol defining the interface for cache backends.

Implementations must provide both synchronous and asynchronous methods for basic cache operations (get, set, delete).

Methods:

Name Description
delete

Synchronously remove an item from cache
adelete

Asynchronously remove an item from cache
get

Synchronously retrieve an item from cache
aget

Asynchronously retrieve an item from cache
set

Synchronously store an item in cache with optional TTL
aset

Asynchronously store an item in cache with optional TTL
Source code in src/pomdapi/core/caching.py 
 8
 9
10
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
class CacheBackend(Protocol):
    """Protocol defining the interface for cache backends.

    Implementations must provide both synchronous and asynchronous methods
    for basic cache operations (get, set, delete).

    Methods:
        delete: Synchronously remove an item from cache
        adelete: Asynchronously remove an item from cache
        get: Synchronously retrieve an item from cache
        aget: Asynchronously retrieve an item from cache
        set: Synchronously store an item in cache with optional TTL
        aset: Asynchronously store an item in cache with optional TTL
    """
    def delete(self, key: str) -> None:
        """Synchronously delete a cache entry by key."""
        ...

    async def adelete(self, key: str) -> None:
        """Asynchronously delete a cache entry by key."""
        ...

    def get(self, key: str) -> Optional[Any]:
        """Synchronously get a cache entry by key."""
        ...

    async def aget(self, key: str) -> Optional[Any]:
        """Asynchronously get a cache entry by key."""
        ...

    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
        """Synchronously set a cache entry with optional TTL in seconds."""
        ...

    async def aset(
        self, key: str, value: Any, ttl: Optional[int] = None
    ) -> None:
        """Asynchronously set a cache entry with optional TTL in seconds."""
        ...

adelete(key) async

Asynchronously delete a cache entry by key.

Source code in src/pomdapi/core/caching.py 
26
27
28
async def adelete(self, key: str) -> None:
    """Asynchronously delete a cache entry by key."""
    ...

aget(key) async

Asynchronously get a cache entry by key.

Source code in src/pomdapi/core/caching.py 
34
35
36
async def aget(self, key: str) -> Optional[Any]:
    """Asynchronously get a cache entry by key."""
    ...

aset(key, value, ttl=None) async

Asynchronously set a cache entry with optional TTL in seconds.

Source code in src/pomdapi/core/caching.py 
42
43
44
45
46
async def aset(
    self, key: str, value: Any, ttl: Optional[int] = None
) -> None:
    """Asynchronously set a cache entry with optional TTL in seconds."""
    ...

delete(key)

Synchronously delete a cache entry by key.

Source code in src/pomdapi/core/caching.py 
22
23
24
def delete(self, key: str) -> None:
    """Synchronously delete a cache entry by key."""
    ...

get(key)

Synchronously get a cache entry by key.

Source code in src/pomdapi/core/caching.py 
30
31
32
def get(self, key: str) -> Optional[Any]:
    """Synchronously get a cache entry by key."""
    ...

set(key, value, ttl=None)

Synchronously set a cache entry with optional TTL in seconds.

Source code in src/pomdapi/core/caching.py 
38
39
40
def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
    """Synchronously set a cache entry with optional TTL in seconds."""
    ...