Skip to content

Extensions

All extensions work with context manager

OnVisitExtension

Runs for each node in the expression

Source code in dynamic_expressions/extensions.py 
 8
 9
10
11
12
13
14
15
16
@runtime_checkable
class OnVisitExtension[Context: EmptyContext](Protocol):
    def on_visit(
        self,
        *,
        node: Node,
        provided_context: Context,
        execution_context: ExecutionContext,
    ) -> AbstractAsyncContextManager[None]: ...

CacheExtension

Caching result for nodes

Base cache extension

Source code in dynamic_expressions/cache/base.py 
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
class CacheExtension[
    Context: EmptyContext,
](OnVisitExtension[Context]):
    policies: Sequence[CachePolicy[Context]]
    default_serializer: Serializer[Any]
    _policy_cache: MutableMapping[type[Node], CachePolicy[Context] | None]

    @contextlib.asynccontextmanager
    async def on_visit(
        self,
        *,
        node: Node,
        provided_context: Context,
        execution_context: ExecutionContext,
    ) -> AsyncIterator[None]:
        policy = self._get_policy(node)

        if policy is None or node in execution_context.cache:
            yield
            return

        key = policy.key(node, provided_context)
        serializer = policy.serializer or self.default_serializer
        cached_value = await self.get_cache(key)
        deserialized_value: object | None = None
        if cached_value is not None:
            deserialized_value = serializer.deserialize(cached_value)
            execution_context.cache[node] = deserialized_value

        yield

        if (
            node in execution_context.cache
            and deserialized_value != execution_context.cache[node]
        ):
            await self.set_cache(
                key=key,
                value=serializer.serialize(execution_context.cache[node]),
                policy=policy,
            )

    def _get_policy(self, node: Node) -> CachePolicy[Context] | None:
        node_cls = type(node)
        if node_cls in self._policy_cache:
            return self._policy_cache[node_cls]

        self._policy_cache[node_cls] = next(
            (policy for policy in self.policies if isinstance(node, policy.types)),
            None,
        )
        return self._policy_cache[node_cls]

    @abc.abstractmethod
    async def get_cache(self, key: str) -> Any | None: ...  # noqa: ANN401

    @abc.abstractmethod
    async def set_cache(
        self,
        key: str,
        value: Any,  # noqa: ANN401
        policy: CachePolicy[Context],
    ) -> None: ...

For redis

Source code in dynamic_expressions/cache/redis.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class RedisCacheExtension[Context: EmptyContext](CacheExtension[Context]):
    def __init__(
        self,
        client: RedisClient,
        policies: Sequence[CachePolicy[Context]],
        default_serializer: Serializer[Any],
    ) -> None:
        self.policies = policies
        self.default_serializer = default_serializer
        self._client = client
        self._policy_cache = {}

    async def get_cache(self, key: str) -> bytes | None:
        return await self._client.get(name=key)

    async def set_cache(
        self,
        key: str,
        value: bytes,
        policy: CachePolicy[Context],
    ) -> None:
        await self._client.set(name=key, value=value, ex=policy.ttl)