fastapi-cache/fastapi_cache/__init__.py at 4542c663906aa8b4d4c8e591495ba88717e6843b · long2ice/fastapi-cache · GitHub

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
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
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
from importlib.metadata import version
from typing import ClassVar, Optional, Type

from fastapi_cache.coder import Coder, JsonCoder
from fastapi_cache.key_builder import default_key_builder
from fastapi_cache.types import Backend, KeyBuilder

__version__ = version("fastapi-cache2")  # pyright: ignore[reportUnknownVariableType]
__all__ = [
    "Backend",
    "Coder",
    "FastAPICache",
    "JsonCoder",
    "KeyBuilder",
    "default_key_builder",
]


class FastAPICache:
    _backend: ClassVar[Optional[Backend]] = None
    _prefix: ClassVar[Optional[str]] = None
    _expire: ClassVar[Optional[int]] = None
    _init: ClassVar[bool] = False
    _coder: ClassVar[Optional[Type[Coder]]] = None
    _key_builder: ClassVar[Optional[KeyBuilder]] = None
    _cache_status_header: ClassVar[Optional[str]] = None
    _enable: ClassVar[bool] = True

    @classmethod
    def init(
        cls,
        backend: Backend,
        prefix: str = "",
        expire: Optional[int] = None,
        coder: Type[Coder] = JsonCoder,
        key_builder: KeyBuilder = default_key_builder,
        cache_status_header: str = "X-FastAPI-Cache",
        enable: bool = True,
    ) -> None:
        if cls._init:
            return
        cls._init = True
        cls._backend = backend
        cls._prefix = prefix
        cls._expire = expire
        cls._coder = coder
        cls._key_builder = key_builder
        cls._cache_status_header = cache_status_header
        cls._enable = enable

    @classmethod
    def reset(cls) -> None:
        cls._init = False
        cls._backend = None
        cls._prefix = None
        cls._expire = None
        cls._coder = None
        cls._key_builder = None
        cls._cache_status_header = None
        cls._enable = True

    @classmethod
    def get_backend(cls) -> Backend:
        assert cls._backend, "You must call init first!"  # noqa: S101
        return cls._backend

    @classmethod
    def get_prefix(cls) -> str:
        assert cls._prefix is not None, "You must call init first!"  # noqa: S101
        return cls._prefix

    @classmethod
    def get_expire(cls) -> Optional[int]:
        return cls._expire

    @classmethod
    def get_coder(cls) -> Type[Coder]:
        assert cls._coder, "You must call init first!"  # noqa: S101
        return cls._coder

    @classmethod
    def get_key_builder(cls) -> KeyBuilder:
        assert cls._key_builder, "You must call init first!"  # noqa: S101
        return cls._key_builder

    @classmethod
    def get_cache_status_header(cls) -> str:
        assert cls._cache_status_header, "You must call init first!"  # noqa: S101
        return cls._cache_status_header

    @classmethod
    def get_enable(cls) -> bool:
        return cls._enable

    @classmethod
    async def clear(
        cls, namespace: Optional[str] = None, key: Optional[str] = None
    ) -> int:
        assert (  # noqa: S101
            cls._backend and cls._prefix is not None
        ), "You must call init first!"
        if namespace is not None or key is None:
            namespace = cls._prefix + (":" + namespace if namespace else "")
        return await cls._backend.clear(namespace, key)