22519085314

Committed 27 Feb 2026 11:47PM UTC coverage: 86.962% (+0.006%) from 86.956%

Build # 22519085314

Build Type

push

github

Committed by

web-flow

Commit Message

SNS: update store typing (#13866)

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

388 existing lines in 19 files now uncovered.

69828 of 80297 relevant lines covered (86.96%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.21

/localstack-core/localstack/services/stepfunctions/asl/eval/variable_store.py

from __future__ import annotations

from typing import Any, Final

from localstack.services.stepfunctions.asl.jsonata.jsonata import (
    VariableDeclarations,
    encode_jsonata_variable_declarations,
)
from localstack.services.stepfunctions.asl.utils.encoding import to_json_str

VariableIdentifier = str
VariableValue = Any


class VariableStoreError(RuntimeError):
    message: Final[str]

    def __init__(self, message: str):
        self.message = message

    def __str__(self):
        return f"{self.__class__.__name__} {self.message}"

    def __repr__(self):
        return str(self)


class NoSuchVariable(VariableStoreError):
    variable_identifier: Final[VariableIdentifier]

    def __init__(self, variable_identifier: VariableIdentifier):
        super().__init__(message=f"No such variable '{variable_identifier}' in scope")
        self.variable_identifier = variable_identifier


class IllegalOuterScopeWrite(VariableStoreError):
    variable_identifier: Final[VariableIdentifier]
    variable_value: Final[VariableValue]

    def __init__(self, variable_identifier: VariableIdentifier, variable_value: VariableValue):
        super().__init__(
            message=f"Cannot bind value '{variable_value}' to variable '{variable_identifier}' as it belongs to an outer scope."
        )
        self.variable_identifier = variable_identifier
        self.variable_value = variable_value


class VariableStore:
    _outer_scope: Final[dict]
    _inner_scope: Final[dict]

    _declaration_tracing: Final[set[str]]

    _outer_variable_declaration_cache: VariableDeclarations | None
    _variable_declarations_cache: VariableDeclarations | None

    def __init__(self, variables: dict | None = None):
        self._outer_scope = {}
        self._inner_scope = {}
        self._declaration_tracing = set()
        self._outer_variable_declaration_cache = None
        self._variable_declarations_cache = None

        if variables:
            for key, value in variables.items():
                self.set(key, value)

    @classmethod
    def as_inner_scope_of(cls, outer_variable_store: VariableStore) -> VariableStore:
        inner_variable_store = cls()
        inner_variable_store._outer_scope.update(outer_variable_store._outer_scope)
        inner_variable_store._outer_scope.update(outer_variable_store._inner_scope)
        return inner_variable_store

    def reset_tracing(self) -> None:
        self._declaration_tracing.clear()

    # TODO: add typing when this available in service init.
    def get_assigned_variables(self) -> dict[str, str]:
        assigned_variables: dict[str, str] = {}
        for traced_declaration_identifier in self._declaration_tracing:
            traced_declaration_value = self.get(traced_declaration_identifier)
            if isinstance(traced_declaration_value, str):
                traced_declaration_value_json_str = f'"{traced_declaration_value}"'
            else:
                traced_declaration_value_json_str: str = to_json_str(
                    traced_declaration_value, separators=(",", ":")
                )
            assigned_variables[traced_declaration_identifier] = traced_declaration_value_json_str
        return assigned_variables

    def to_dict(self) -> dict[str, str]:
        assigned_variables: dict[str, str] = {}
        for traced_declaration_identifier in self._declaration_tracing:
            assigned_variables[traced_declaration_identifier] = self.get(
                traced_declaration_identifier
            )
        return assigned_variables

    def get(self, variable_identifier: VariableIdentifier) -> VariableValue:
        if variable_identifier in self._inner_scope:
            return self._inner_scope[variable_identifier]
        if variable_identifier in self._outer_scope:
            return self._outer_scope[variable_identifier]
        raise NoSuchVariable(variable_identifier=variable_identifier)

    def set(self, variable_identifier: VariableIdentifier, variable_value: VariableValue) -> None:
        if variable_identifier in self._outer_scope:
            raise IllegalOuterScopeWrite(
                variable_identifier=variable_identifier, variable_value=variable_value
            )
        self._declaration_tracing.add(variable_identifier)
        self._inner_scope[variable_identifier] = variable_value
        self._variable_declarations_cache = None

    @staticmethod
    def _to_variable_declarations(bindings: dict[str, Any]) -> VariableDeclarations:
        variables = {f"${key}": value for key, value in bindings.items()}
        encoded = encode_jsonata_variable_declarations(variables)
        return encoded

    def get_variable_declarations(self) -> VariableDeclarations:
        if self._variable_declarations_cache is not None:
            return self._variable_declarations_cache
        if self._outer_variable_declaration_cache is None:
            self._outer_variable_declaration_cache = self._to_variable_declarations(
                self._outer_scope
            )
        inner_variable_declarations_cache = self._to_variable_declarations(self._inner_scope)
        self._variable_declarations_cache = "".join(
            [self._outer_variable_declaration_cache, inner_variable_declarations_cache]
        )
        return self._variable_declarations_cache

1	from __future__ import annotations	1✔
2
3	from typing import Any, Final	1✔
4
5	from localstack.services.stepfunctions.asl.jsonata.jsonata import (	1✔
6	VariableDeclarations,
7	encode_jsonata_variable_declarations,
8	)
9	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
10
11	VariableIdentifier = str	1✔
12	VariableValue = Any	1✔
13
14
15	class VariableStoreError(RuntimeError):	1✔
16	message: Final[str]	1✔
17
18	def __init__(self, message: str):	1✔
19	self.message = message	×
20
21	def __str__(self):	1✔
22	return f"{self.__class__.__name__} {self.message}"	×
23
24	def __repr__(self):
25	return str(self)
26
27
28	class NoSuchVariable(VariableStoreError):	1✔
29	variable_identifier: Final[VariableIdentifier]	1✔
30
31	def __init__(self, variable_identifier: VariableIdentifier):	1✔
32	super().__init__(message=f"No such variable '{variable_identifier}' in scope")	×
33	self.variable_identifier = variable_identifier	×
34
35
36	class IllegalOuterScopeWrite(VariableStoreError):	1✔
37	variable_identifier: Final[VariableIdentifier]	1✔
38	variable_value: Final[VariableValue]	1✔
39
40	def __init__(self, variable_identifier: VariableIdentifier, variable_value: VariableValue):	1✔
41	super().__init__(	×
42	message=f"Cannot bind value '{variable_value}' to variable '{variable_identifier}' as it belongs to an outer scope."
43	)
44	self.variable_identifier = variable_identifier	×
45	self.variable_value = variable_value	×
46
47
48	class VariableStore:	1✔
49	_outer_scope: Final[dict]	1✔
50	_inner_scope: Final[dict]	1✔
51
52	_declaration_tracing: Final[set[str]]	1✔
53
54	_outer_variable_declaration_cache: VariableDeclarations \| None	1✔
55	_variable_declarations_cache: VariableDeclarations \| None	1✔
56
57	def __init__(self, variables: dict \| None = None):	1✔
58	self._outer_scope = {}	1✔
59	self._inner_scope = {}	1✔
60	self._declaration_tracing = set()	1✔
61	self._outer_variable_declaration_cache = None	1✔
62	self._variable_declarations_cache = None	1✔
63
64	if variables:	1✔
65	for key, value in variables.items():	1✔
66	self.set(key, value)	1✔
67
68	@classmethod	1✔
69	def as_inner_scope_of(cls, outer_variable_store: VariableStore) -> VariableStore:	1✔
70	inner_variable_store = cls()	1✔
71	inner_variable_store._outer_scope.update(outer_variable_store._outer_scope)	1✔
72	inner_variable_store._outer_scope.update(outer_variable_store._inner_scope)	1✔
73	return inner_variable_store	1✔
74
75	def reset_tracing(self) -> None:	1✔
76	self._declaration_tracing.clear()	1✔
77
78	# TODO: add typing when this available in service init.
79	def get_assigned_variables(self) -> dict[str, str]:	1✔
80	assigned_variables: dict[str, str] = {}	1✔
81	for traced_declaration_identifier in self._declaration_tracing:	1✔
82	traced_declaration_value = self.get(traced_declaration_identifier)	1✔
83	if isinstance(traced_declaration_value, str):	1✔
84	traced_declaration_value_json_str = f'"{traced_declaration_value}"'	1✔
85	else:
86	traced_declaration_value_json_str: str = to_json_str(	1✔
87	traced_declaration_value, separators=(",", ":")
88	)
89	assigned_variables[traced_declaration_identifier] = traced_declaration_value_json_str	1✔
90	return assigned_variables	1✔
91
92	def to_dict(self) -> dict[str, str]:	1✔
93	assigned_variables: dict[str, str] = {}	1✔
94	for traced_declaration_identifier in self._declaration_tracing:	1✔
95	assigned_variables[traced_declaration_identifier] = self.get(	1✔
96	traced_declaration_identifier
97	)
98	return assigned_variables	1✔
99
100	def get(self, variable_identifier: VariableIdentifier) -> VariableValue:	1✔
101	if variable_identifier in self._inner_scope:	1✔
102	return self._inner_scope[variable_identifier]	1✔
UNCOV 103	if variable_identifier in self._outer_scope:	×
UNCOV 104	return self._outer_scope[variable_identifier]	×
UNCOV 105	raise NoSuchVariable(variable_identifier=variable_identifier)	×
106
107	def set(self, variable_identifier: VariableIdentifier, variable_value: VariableValue) -> None:	1✔
108	if variable_identifier in self._outer_scope:	1✔
UNCOV 109	raise IllegalOuterScopeWrite(	×
110	variable_identifier=variable_identifier, variable_value=variable_value
111	)
112	self._declaration_tracing.add(variable_identifier)	1✔
113	self._inner_scope[variable_identifier] = variable_value	1✔
114	self._variable_declarations_cache = None	1✔
115
116	@staticmethod	1✔
117	def _to_variable_declarations(bindings: dict[str, Any]) -> VariableDeclarations:	1✔
118	variables = {f"${key}": value for key, value in bindings.items()}	1✔
119	encoded = encode_jsonata_variable_declarations(variables)	1✔
120	return encoded	1✔
121
122	def get_variable_declarations(self) -> VariableDeclarations:	1✔
123	if self._variable_declarations_cache is not None:	1✔
124	return self._variable_declarations_cache	1✔
125	if self._outer_variable_declaration_cache is None:	1✔
126	self._outer_variable_declaration_cache = self._to_variable_declarations(	1✔
127	self._outer_scope
128	)
129	inner_variable_declarations_cache = self._to_variable_declarations(self._inner_scope)	1✔
130	self._variable_declarations_cache = "".join(	1✔
131	[self._outer_variable_declaration_cache, inner_variable_declarations_cache]
132	)
133	return self._variable_declarations_cache	1✔

localstack / localstack / 22519085314

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous