1
0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-11-22 23:48:12 +00:00

fixup! docs(core): add and modify docs to context and cache [no changelog]

This commit is contained in:
M1nd3r 2024-11-22 11:46:55 +01:00
parent 35856917c3
commit 7f11fea37f
5 changed files with 53 additions and 11 deletions

View File

@ -11,6 +11,16 @@ if TYPE_CHECKING:
def stored(key: int) -> Callable[[ByteFunc[P]], ByteFunc[P]]:
"""
Caches the result of a function call based on the given key.
- If the key is already present in the cache, the cached value is returned
directly without invoking the decorated function.
- If the key is not present in the cache, the decorated function is executed,
and its result is stored in the cache before being returned to the caller.
"""
def decorator(func: ByteFunc[P]) -> ByteFunc[P]:
def wrapper(*args: P.args, **kwargs: P.kwargs) -> bytes:
@ -26,6 +36,17 @@ def stored(key: int) -> Callable[[ByteFunc[P]], ByteFunc[P]]:
def stored_async(key: int) -> Callable[[AsyncByteFunc[P]], AsyncByteFunc[P]]:
"""
Caches the result of an async function call based on the given key.
- If the key is already present in the cache, the cached value is returned
directly without invoking the decorated asynchronous function.
- If the key is not present in the cache, the decorated asynchronous function
is executed, and its result is stored in the cache before being returned
to the caller.
"""
def decorator(func: AsyncByteFunc[P]) -> AsyncByteFunc[P]:
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> bytes:
value = context.cache_get(key)

View File

@ -4,7 +4,6 @@ import gc
from storage import cache_codec
from storage.cache_common import SESSIONLESS_FLAG, SessionlessCache
# Cache initialization
_SESSIONLESS_CACHE = SessionlessCache()
_PROTOCOL_CACHE = cache_codec
@ -15,6 +14,9 @@ gc.collect()
def clear_all() -> None:
"""
Clears all data from both the protocol cache and the sessionless cache.
"""
global autolock_last_touch
autolock_last_touch = None
_SESSIONLESS_CACHE.clear()
@ -22,6 +24,13 @@ def clear_all() -> None:
def get_int_all_sessions(key: int) -> builtins.set[int]:
"""
Returns set of int values associated with a given key from all relevant sessions.
If the key has the `SESSIONLESS_FLAG` set, the values are retrieved
from the sessionless cache. Otherwise, the values are fetched
from the protocol cache.
"""
if key & SESSIONLESS_FLAG:
values = builtins.set()
encoded = _SESSIONLESS_CACHE.get(key)

View File

@ -129,6 +129,7 @@ if __debug__:
mem_info(True)
def get_bytes_as_str(a: bytes) -> str:
"""Converts the provided bytes to a hexadecimal string (decoded as`utf-8`)."""
return hexlify(a).decode("utf-8")

View File

@ -35,12 +35,6 @@ class CodecContext(Context):
expected_types: Container[int],
expected_type: type[protobuf.MessageType] | None = None,
) -> protobuf.MessageType:
"""Read a message from the wire.
The read message must be of one of the types specified in `expected_types`.
If only a single type is expected, it can be passed as `expected_type`,
to save on having to decode the type code into a protobuf class.
"""
if __debug__:
log.debug(
__name__,
@ -74,7 +68,6 @@ class CodecContext(Context):
return wrap_protobuf_load(msg.data, expected_type)
async def write(self, msg: protobuf.MessageType) -> None:
"""Write a message to the wire."""
if __debug__:
log.debug(
__name__,

View File

@ -13,6 +13,11 @@ if TYPE_CHECKING:
class Message:
"""
Encapsulates protobuf encoded message, where
- `type` is the `WIRE_TYPE` of the message
- `data` is the protobuf encoded message
"""
def __init__(
self,
@ -54,15 +59,25 @@ class Context:
self,
expected_types: Container[int],
expected_type: type[protobuf.MessageType] | None = None,
) -> protobuf.MessageType: ...
) -> protobuf.MessageType:
"""Read a message from the wire.
async def write(self, msg: protobuf.MessageType) -> None: ...
The read message must be of one of the types specified in `expected_types`.
If only a single type is expected, it can be passed as `expected_type`,
to save on having to decode the type code into a protobuf class.
"""
...
async def write(self, msg: protobuf.MessageType) -> None:
"""Write a message to the wire."""
...
async def call(
self,
msg: protobuf.MessageType,
expected_type: type[LoadedMessageType],
) -> LoadedMessageType:
"""Write a message to the wire, then await and return the response message."""
assert expected_type.MESSAGE_WIRE_TYPE is not None
await self.write(msg)
@ -70,10 +85,13 @@ class Context:
return await self.read((expected_type.MESSAGE_WIRE_TYPE,), expected_type)
def release(self) -> None:
"""Release resources used by the context, eg. clear context cache."""
pass
@property
def cache(self) -> DataCache: ...
def cache(self) -> DataCache:
"""Access to the backing cache of the context, if the context has any."""
...
class WireError(Exception):