21575916503

Committed 30 Jan 2026 10:27AM UTC coverage: 86.969% (+0.007%) from 86.962%

Build # 21575916503

Build Type

push

github

Committed by

web-flow

Commit Message

Admin: Add typehints to utils/strings and utils/threads (#13658)

Run Details

39 of 42 new or added lines in 3 files covered. (92.86%)

27 existing lines in 1 file now uncovered.

70391 of 80938 relevant lines covered (86.97%)

0.87 hits per line

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

89.92

/localstack-core/localstack/utils/strings.py

import base64
import binascii
import hashlib
import itertools
import random
import re
import string
import uuid
import zlib
from typing import TYPE_CHECKING, Any

from localstack.config import DEFAULT_ENCODING

if TYPE_CHECKING:
    from localstack.utils.objects import ComplexType

_unprintables = (
    range(0x00, 0x09),
    range(0x0A, 0x0A),
    range(0x0B, 0x0D),
    range(0x0E, 0x20),
    range(0xD800, 0xE000),
    range(0xFFFE, 0x10000),
)

# regular expression for unprintable characters
# Based on https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html
#     #x9 | #xA | #xD | #x20 to #xD7FF | #xE000 to #xFFFD | #x10000 to #x10FFFF
REGEX_UNPRINTABLE_CHARS = re.compile(
    f"[{re.escape(''.join(map(chr, itertools.chain(*_unprintables))))}]"
)


def to_str(obj: str | bytes, encoding: str = DEFAULT_ENCODING, errors: str = "strict") -> str:
    """If ``obj`` is an instance of ``binary_type``, return
    ``obj.decode(encoding, errors)``, otherwise return ``obj``"""
    return obj.decode(encoding, errors) if isinstance(obj, bytes) else obj


def to_bytes(obj: str | bytes, encoding: str = DEFAULT_ENCODING, errors: str = "strict") -> bytes:
    """If ``obj`` is an instance of ``text_type``, return
    ``obj.encode(encoding, errors)``, otherwise return ``obj``"""
    return obj.encode(encoding, errors) if isinstance(obj, str) else obj


def truncate(data: str, max_length: int = 100) -> str:
    data = str(data or "")
    return (f"{data[:max_length]}...") if len(data) > max_length else data


def is_string(s: Any, include_unicode: bool = True, exclude_binary: bool = False) -> bool:
    if isinstance(s, bytes) and exclude_binary:
        return False
    if isinstance(s, str):
        return True
    if include_unicode and isinstance(s, str):
        return True
    return False


def is_string_or_bytes(s: Any) -> bool:
    return is_string(s) or isinstance(s, str) or isinstance(s, bytes)


def is_base64(s: Any) -> bool:
    regex = r"^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$"
    return is_string(s) and re.match(regex, s) is not None


_re_camel_to_snake_case = re.compile("((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))")


def camel_to_snake_case(string: str) -> str:
    return _re_camel_to_snake_case.sub(r"_\1", string).replace("__", "_").lower()


def snake_to_camel_case(string: str, capitalize_first: bool = True) -> str:
    components = string.split("_")
    start_idx = 0 if capitalize_first else 1
    components = [x.title() for x in components[start_idx:]]
    return "".join(components)


def hyphen_to_snake_case(string: str) -> str:
    return string.replace("-", "_")


def canonicalize_bool_to_str(val: bool) -> str:
    return "true" if str(val).lower() == "true" else "false"


def convert_to_printable_chars(value: "str | ComplexType") -> "ComplexType":
    """Removes all unprintable characters from the given string."""
    from localstack.utils.objects import recurse_object

    if isinstance(value, str):
        return REGEX_UNPRINTABLE_CHARS.sub("", value)
    else:

        def _convert(obj: Any, **kwargs: Any) -> "ComplexType":
            if isinstance(obj, str):
                return convert_to_printable_chars(obj)
            return obj

        return recurse_object(value, _convert)


def first_char_to_lower(s: str) -> str:
    return s and f"{s[0].lower()}{s[1:]}"


def first_char_to_upper(s: str) -> str:
    return s and f"{s[0].upper()}{s[1:]}"


def str_to_bool(value: Any) -> Any:
    """Return the boolean value of the given string, or the verbatim value if it is not a string"""
    if isinstance(value, str):
        true_strings = ["true", "True"]
        return value in true_strings
    return value


def str_insert(string: str, index: int, content: str) -> str:
    """Insert a substring into an existing string at a certain index."""
    return f"{string[:index]}{content}{string[index:]}"


def str_remove(string: str, index: int, end_index: int | None = None) -> str:
    """Remove a substring from an existing string at a certain from-to index range."""
    end_index = end_index or (index + 1)
    return f"{string[:index]}{string[end_index:]}"


def str_startswith_ignore_case(value: str, prefix: str) -> bool:
    return value[: len(prefix)].lower() == prefix.lower()


def short_uid() -> str:
    return str(uuid.uuid4())[0:8]


def short_uid_from_seed(seed: str) -> str:
    hash = hashlib.sha1(seed.encode("utf-8")).hexdigest()
    truncated_hash = hash[:32]
    return str(uuid.UUID(truncated_hash))[0:8]


def long_uid() -> str:
    return str(uuid.uuid4())


def md5(string: str | bytes) -> str:
    m = hashlib.md5()
    m.update(to_bytes(string))
    return m.hexdigest()


def checksum_crc32(string: str | bytes) -> str:
    bytes = to_bytes(string)
    checksum = zlib.crc32(bytes)
    return base64.b64encode(checksum.to_bytes(4, "big")).decode()


def checksum_crc32c(string: str | bytes) -> str:
    # import botocore locally here to avoid a dependency of the CLI to botocore
    from botocore.httpchecksum import CrtCrc32cChecksum

    checksum = CrtCrc32cChecksum()
    checksum.update(to_bytes(string))
    return base64.b64encode(checksum.digest()).decode()


def checksum_crc64nvme(string: str | bytes) -> str:
    # import botocore locally here to avoid a dependency of the CLI to botocore
    from botocore.httpchecksum import CrtCrc64NvmeChecksum

    checksum = CrtCrc64NvmeChecksum()
    checksum.update(to_bytes(string))
    return base64.b64encode(checksum.digest()).decode()


def hash_sha1(string: str | bytes) -> str:
    digest = hashlib.sha1(to_bytes(string)).digest()
    return base64.b64encode(digest).decode()


def hash_sha256(string: str | bytes) -> str:
    digest = hashlib.sha256(to_bytes(string)).digest()
    return base64.b64encode(digest).decode()


def base64_to_hex(b64_string: str) -> bytes:
    return binascii.hexlify(base64.b64decode(b64_string))


def base64_decode(data: str | bytes) -> bytes:
    """Decode base64 data - with optional padding, and able to handle urlsafe encoding (containing -/_)."""
    data = to_str(data)
    missing_padding = len(data) % 4
    if missing_padding != 0:
        data = to_str(data) + "=" * (4 - missing_padding)
    if "-" in data or "_" in data:
        return base64.urlsafe_b64decode(data)
    return base64.b64decode(data)


def get_random_hex(length: int) -> str:
    return "".join(random.choices(string.hexdigits[:16], k=length)).lower()


def remove_leading_extra_slashes(input: str) -> str:
    """
    Remove leading extra slashes from the given input string.
    Example: '///foo/bar' -> '/foo/bar'
    """
    return re.sub(r"^/+", "/", input)


def prepend_with_slash(input: str) -> str:
    """
    Prepend a slash `/` to a given string if it does not have one already.
    """
    if not input.startswith("/"):
        return f"/{input}"
    return input


def key_value_pairs_to_dict(
    pairs: str, delimiter: str = ",", separator: str = "="
) -> dict[str, str]:
    """
    Converts a string of key-value pairs to a dictionary.

    Args:
        pairs (str): A string containing key-value pairs separated by a delimiter.
        delimiter (str): The delimiter used to separate key-value pairs (default is comma ',').
        separator (str): The separator between keys and values (default is '=').

    Returns:
        dict: A dictionary containing the parsed key-value pairs.
    """
    splits = [split_pair.partition(separator) for split_pair in pairs.split(delimiter)]
    return {key.strip(): value.strip() for key, _, value in splits}


def token_generator(item: str) -> str:
    base64_bytes = base64.b64encode(item.encode("utf-8"))
    token = base64_bytes.decode("utf-8")
    return token

1	import base64	1✔
2	import binascii	1✔
3	import hashlib	1✔
4	import itertools	1✔
5	import random	1✔
6	import re	1✔
7	import string	1✔
8	import uuid	1✔
9	import zlib	1✔
10	from typing import TYPE_CHECKING, Any	1✔
11
12	from localstack.config import DEFAULT_ENCODING	1✔
13
14	if TYPE_CHECKING:	1✔
NEW 15	from localstack.utils.objects import ComplexType	×
16
17	_unprintables = (	1✔
18	range(0x00, 0x09),
19	range(0x0A, 0x0A),
20	range(0x0B, 0x0D),
21	range(0x0E, 0x20),
22	range(0xD800, 0xE000),
23	range(0xFFFE, 0x10000),
24	)
25
26	# regular expression for unprintable characters
27	# Based on https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html
28	# #x9 \| #xA \| #xD \| #x20 to #xD7FF \| #xE000 to #xFFFD \| #x10000 to #x10FFFF
29	REGEX_UNPRINTABLE_CHARS = re.compile(	1✔
30	f"[{re.escape(''.join(map(chr, itertools.chain(*_unprintables))))}]"
31	)
32
33
34	def to_str(obj: str \| bytes, encoding: str = DEFAULT_ENCODING, errors: str = "strict") -> str:	1✔
35	"""If ``obj`` is an instance of ``binary_type``, return
36	``obj.decode(encoding, errors)``, otherwise return ``obj``"""
37	return obj.decode(encoding, errors) if isinstance(obj, bytes) else obj	1✔
38
39
40	def to_bytes(obj: str \| bytes, encoding: str = DEFAULT_ENCODING, errors: str = "strict") -> bytes:	1✔
41	"""If ``obj`` is an instance of ``text_type``, return
42	``obj.encode(encoding, errors)``, otherwise return ``obj``"""
43	return obj.encode(encoding, errors) if isinstance(obj, str) else obj	1✔
44
45
46	def truncate(data: str, max_length: int = 100) -> str:	1✔
47	data = str(data or "")	1✔
48	return (f"{data[:max_length]}...") if len(data) > max_length else data	1✔
49
50
51	def is_string(s: Any, include_unicode: bool = True, exclude_binary: bool = False) -> bool:	1✔
52	if isinstance(s, bytes) and exclude_binary:	1✔
53	return False	×
54	if isinstance(s, str):	1✔
55	return True	1✔
56	if include_unicode and isinstance(s, str):	×
57	return True	×
58	return False	×
59
60
61	def is_string_or_bytes(s: Any) -> bool:	1✔
62	return is_string(s) or isinstance(s, str) or isinstance(s, bytes)	×
63
64
65	def is_base64(s: Any) -> bool:	1✔
66	regex = r"^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==\|[A-Za-z0-9+/]{3}=)?$"	1✔
67	return is_string(s) and re.match(regex, s) is not None	1✔
68
69
70	_re_camel_to_snake_case = re.compile("((?<=[a-z0-9])[A-Z]\|(?!^)[A-Z](?=[a-z]))")	1✔
71
72
73	def camel_to_snake_case(string: str) -> str:	1✔
74	return _re_camel_to_snake_case.sub(r"_\1", string).replace("__", "_").lower()	1✔
75
76
77	def snake_to_camel_case(string: str, capitalize_first: bool = True) -> str:	1✔
78	components = string.split("_")	1✔
79	start_idx = 0 if capitalize_first else 1	1✔
80	components = [x.title() for x in components[start_idx:]]	1✔
81	return "".join(components)	1✔
82
83
84	def hyphen_to_snake_case(string: str) -> str:	1✔
85	return string.replace("-", "_")	×
86
87
88	def canonicalize_bool_to_str(val: bool) -> str:	1✔
89	return "true" if str(val).lower() == "true" else "false"	1✔
90
91
92	def convert_to_printable_chars(value: "str \| ComplexType") -> "ComplexType":	1✔
93	"""Removes all unprintable characters from the given string."""
94	from localstack.utils.objects import recurse_object	1✔
95
96	if isinstance(value, str):	1✔
97	return REGEX_UNPRINTABLE_CHARS.sub("", value)	1✔
98	else:
99
100	def _convert(obj: Any, **kwargs: Any) -> "ComplexType":	1✔
101	if isinstance(obj, str):	1✔
102	return convert_to_printable_chars(obj)	1✔
103	return obj	1✔
104
105	return recurse_object(value, _convert)	1✔
106
107
108	def first_char_to_lower(s: str) -> str:	1✔
109	return s and f"{s[0].lower()}{s[1:]}"	1✔
110
111
112	def first_char_to_upper(s: str) -> str:	1✔
113	return s and f"{s[0].upper()}{s[1:]}"	1✔
114
115
116	def str_to_bool(value: Any) -> Any:	1✔
117	"""Return the boolean value of the given string, or the verbatim value if it is not a string"""
118	if isinstance(value, str):	1✔
119	true_strings = ["true", "True"]	1✔
120	return value in true_strings	1✔
121	return value	1✔
122
123
124	def str_insert(string: str, index: int, content: str) -> str:	1✔
125	"""Insert a substring into an existing string at a certain index."""
126	return f"{string[:index]}{content}{string[index:]}"	×
127
128
129	def str_remove(string: str, index: int, end_index: int \| None = None) -> str:	1✔
130	"""Remove a substring from an existing string at a certain from-to index range."""
131	end_index = end_index or (index + 1)	×
132	return f"{string[:index]}{string[end_index:]}"	×
133
134
135	def str_startswith_ignore_case(value: str, prefix: str) -> bool:	1✔
136	return value[: len(prefix)].lower() == prefix.lower()	×
137
138
139	def short_uid() -> str:	1✔
140	return str(uuid.uuid4())[0:8]	1✔
141
142
143	def short_uid_from_seed(seed: str) -> str:	1✔
144	hash = hashlib.sha1(seed.encode("utf-8")).hexdigest()	1✔
145	truncated_hash = hash[:32]	1✔
146	return str(uuid.UUID(truncated_hash))[0:8]	1✔
147
148
149	def long_uid() -> str:	1✔
150	return str(uuid.uuid4())	1✔
151
152
153	def md5(string: str \| bytes) -> str:	1✔
154	m = hashlib.md5()	1✔
155	m.update(to_bytes(string))	1✔
156	return m.hexdigest()	1✔
157
158
159	def checksum_crc32(string: str \| bytes) -> str:	1✔
160	bytes = to_bytes(string)	1✔
161	checksum = zlib.crc32(bytes)	1✔
162	return base64.b64encode(checksum.to_bytes(4, "big")).decode()	1✔
163
164
165	def checksum_crc32c(string: str \| bytes) -> str:	1✔
166	# import botocore locally here to avoid a dependency of the CLI to botocore
167	from botocore.httpchecksum import CrtCrc32cChecksum	1✔
168
169	checksum = CrtCrc32cChecksum()	1✔
170	checksum.update(to_bytes(string))	1✔
171	return base64.b64encode(checksum.digest()).decode()	1✔
172
173
174	def checksum_crc64nvme(string: str \| bytes) -> str:	1✔
175	# import botocore locally here to avoid a dependency of the CLI to botocore
176	from botocore.httpchecksum import CrtCrc64NvmeChecksum	1✔
177
178	checksum = CrtCrc64NvmeChecksum()	1✔
179	checksum.update(to_bytes(string))	1✔
180	return base64.b64encode(checksum.digest()).decode()	1✔
181
182
183	def hash_sha1(string: str \| bytes) -> str:	1✔
184	digest = hashlib.sha1(to_bytes(string)).digest()	1✔
185	return base64.b64encode(digest).decode()	1✔
186
187
188	def hash_sha256(string: str \| bytes) -> str:	1✔
189	digest = hashlib.sha256(to_bytes(string)).digest()	1✔
190	return base64.b64encode(digest).decode()	1✔
191
192
193	def base64_to_hex(b64_string: str) -> bytes:	1✔
194	return binascii.hexlify(base64.b64decode(b64_string))	1✔
195
196
197	def base64_decode(data: str \| bytes) -> bytes:	1✔
198	"""Decode base64 data - with optional padding, and able to handle urlsafe encoding (containing -/_)."""
199	data = to_str(data)	1✔
200	missing_padding = len(data) % 4	1✔
201	if missing_padding != 0:	1✔
202	data = to_str(data) + "=" * (4 - missing_padding)	×
203	if "-" in data or "_" in data:	1✔
204	return base64.urlsafe_b64decode(data)	1✔
205	return base64.b64decode(data)	1✔
206
207
208	def get_random_hex(length: int) -> str:	1✔
209	return "".join(random.choices(string.hexdigits[:16], k=length)).lower()	1✔
210
211
212	def remove_leading_extra_slashes(input: str) -> str:	1✔
213	"""
214	Remove leading extra slashes from the given input string.
215	Example: '///foo/bar' -> '/foo/bar'
216	"""
217	return re.sub(r"^/+", "/", input)	×
218
219
220	def prepend_with_slash(input: str) -> str:	1✔
221	"""
222	Prepend a slash `/` to a given string if it does not have one already.
223	"""
224	if not input.startswith("/"):	1✔
225	return f"/{input}"	1✔
226	return input	1✔
227
228
229	def key_value_pairs_to_dict(	1✔
230	pairs: str, delimiter: str = ",", separator: str = "="
231	) -> dict[str, str]:
232	"""
233	Converts a string of key-value pairs to a dictionary.
234
235	Args:
236	pairs (str): A string containing key-value pairs separated by a delimiter.
237	delimiter (str): The delimiter used to separate key-value pairs (default is comma ',').
238	separator (str): The separator between keys and values (default is '=').
239
240	Returns:
241	dict: A dictionary containing the parsed key-value pairs.
242	"""
243	splits = [split_pair.partition(separator) for split_pair in pairs.split(delimiter)]	1✔
244	return {key.strip(): value.strip() for key, _, value in splits}	1✔
245
246
247	def token_generator(item: str) -> str:	1✔
248	base64_bytes = base64.b64encode(item.encode("utf-8"))	1✔
249	token = base64_bytes.decode("utf-8")	1✔
250	return token	1✔

localstack / localstack / 21575916503

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