File: cache.py

package info (click to toggle)
python-beanie 2.0.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,496 kB
  • sloc: python: 14,596; makefile: 6; sh: 6
file content (45 lines) | stat: -rw-r--r-- 1,316 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
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
import collections
import datetime
from datetime import timedelta, timezone
from typing import Any, Optional

from pydantic import BaseModel, Field


class CachedItem(BaseModel):
    timestamp: datetime.datetime = Field(
        default_factory=lambda: datetime.datetime.now(tz=timezone.utc)
    )
    value: Any


class LRUCache:
    def __init__(self, capacity: int, expiration_time: timedelta):
        self.capacity: int = capacity
        self.expiration_time: timedelta = expiration_time
        self.cache: collections.OrderedDict = collections.OrderedDict()

    def get(self, key) -> Optional[CachedItem]:
        try:
            item: CachedItem = self.cache.pop(key)
            if (
                datetime.datetime.now(tz=timezone.utc) - item.timestamp
                > self.expiration_time
            ):
                return None
            self.cache[key] = item
            return item.value
        except KeyError:
            return None

    def set(self, key, value) -> None:
        try:
            self.cache.pop(key)
        except KeyError:
            if len(self.cache) >= self.capacity:
                self.cache.popitem(last=False)
        self.cache[key] = CachedItem(value=value)

    @staticmethod
    def create_key(*args):
        return str(args)  # TODO think about this