16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

92.55

/localstack-core/localstack/aws/spec.py

import dataclasses
import json
import logging
import os
import sys
from collections import defaultdict
from collections.abc import Generator
from functools import cached_property, lru_cache
from typing import Literal, NamedTuple

import botocore
import jsonpatch
from botocore.exceptions import UnknownServiceError
from botocore.loaders import Loader, instance_cache
from botocore.model import OperationModel, ServiceModel

from localstack import config
from localstack.constants import VERSION
from localstack.utils.objects import singleton_factory

LOG = logging.getLogger(__name__)

ServiceName = str
ProtocolName = Literal["query", "json", "rest-json", "rest-xml", "ec2"]


class ServiceModelIdentifier(NamedTuple):
    """
    Identifies a specific service model.
    If the protocol is not given, the default protocol of the service with the specific name is assumed.
    Maybe also add versions here in the future (if we can support multiple different versions for one service).
    """

    name: ServiceName
    protocol: ProtocolName | None = None


spec_patches_json = os.path.join(os.path.dirname(__file__), "spec-patches.json")


def load_spec_patches() -> dict[str, list]:
    if not os.path.exists(spec_patches_json):
        return {}
    with open(spec_patches_json) as fd:
        return json.load(fd)


# Path for custom specs which are not (anymore) provided by botocore
LOCALSTACK_BUILTIN_DATA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")


class LocalStackBuiltInDataLoaderMixin(Loader):
    def __init__(self, *args, **kwargs):
        # add the builtin data path to the extra_search_paths to ensure they are discovered by the loader
        super().__init__(*args, extra_search_paths=[LOCALSTACK_BUILTIN_DATA_PATH], **kwargs)


class PatchingLoader(Loader):
    """
    A custom botocore Loader that applies JSON patches from the given json patch file to the specs as they are loaded.
    """

    patches: dict[str, list]

    def __init__(self, patches: dict[str, list], *args, **kwargs):
        # add the builtin data path to the extra_search_paths to ensure they are discovered by the loader
        super().__init__(*args, **kwargs)
        self.patches = patches

    @instance_cache
    def load_data(self, name: str):
        result = super(PatchingLoader, self).load_data(name)

        if patches := self.patches.get(name):
            return jsonpatch.apply_patch(result, patches)

        return result


class CustomLoader(PatchingLoader, LocalStackBuiltInDataLoaderMixin):
    # Class mixing the different loader features (patching, localstack specific data)
    pass


loader = CustomLoader(load_spec_patches())


class UnknownServiceProtocolError(UnknownServiceError):
    """Raised when trying to load a service with an unknown protocol.

    :ivar service_name: The name of the service.
    :ivar protocol: The name of the unknown protocol.
    """

    fmt = "Unknown service protocol: '{service_name}-{protocol}'."


def list_services() -> list[ServiceModel]:
    return [load_service(service) for service in loader.list_available_services("service-2")]


def load_service(
    service: ServiceName, version: str | None = None, protocol: ProtocolName | None = None
) -> ServiceModel:
    """
    Loads a service
    :param service: to load, f.e. "sqs". For custom, internalized, service protocol specs (f.e. sqs-query) it's also
                    possible to directly define the protocol in the service name (f.e. use sqs-query)
    :param version: of the service to load, f.e. "2012-11-05", by default the latest version will be used
    :param protocol: specific protocol to load for the specific service, f.e. "json" for the "sqs" service
                     if the service cannot be found
    :return: Loaded service model of the service
    :raises: UnknownServiceError if the service cannot be found
    :raises: UnknownServiceProtocolError if the specific protocol of the service cannot be found
    """
    service_description = loader.load_service_model(service, "service-2", version)

    # check if the protocol is defined, and if so, if the loaded service defines this protocol
    if protocol is not None and protocol != service_description.get("metadata", {}).get("protocol"):
        # if the protocol is defined, but not the one of the currently loaded service,
        # check if we already loaded the custom spec based on the naming convention (<service>-<protocol>),
        # f.e. "sqs-query"
        if service.endswith(f"-{protocol}"):
            # if so, we raise an exception
            raise UnknownServiceProtocolError(service_name=service, protocol=protocol)
        # otherwise we try to load it (recursively)
        try:
            return load_service(f"{service}-{protocol}", version, protocol=protocol)
        except UnknownServiceError:
            # raise an unknown protocol error in case the service also can't be loaded with the naming convention
            raise UnknownServiceProtocolError(service_name=service, protocol=protocol)

    # remove potential protocol names from the service name
    # FIXME add more protocols here if we have to internalize more than just sqs-query
    # TODO this should not contain specific internalized serivce names
    service = {"sqs-query": "sqs"}.get(service, service)
    return ServiceModel(service_description, service)


def iterate_service_operations() -> Generator[tuple[ServiceModel, OperationModel], None, None]:
    """
    Returns one record per operation in the AWS service spec, where the first item is the service model the operation
    belongs to, and the second is the operation model.

    :return: an iterable
    """
    for service in list_services():
        for op_name in service.operation_names:
            yield service, service.operation_model(op_name)


@dataclasses.dataclass
class ServiceCatalogIndex:
    """
    The ServiceCatalogIndex enables fast lookups for common operations to determine a service from service indicators.
    """

    service_names: list[ServiceName]
    target_prefix_index: dict[str, list[ServiceModelIdentifier]]
    signing_name_index: dict[str, list[ServiceModelIdentifier]]
    operations_index: dict[str, list[ServiceModelIdentifier]]
    endpoint_prefix_index: dict[str, list[ServiceModelIdentifier]]


class LazyServiceCatalogIndex:
    """
    A ServiceCatalogIndex that builds indexes in-memory from the spec.
    """

    @cached_property
    def service_names(self) -> list[ServiceName]:
        return list(self._services.keys())

    @cached_property
    def target_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        result = defaultdict(list)
        for service_models in self._services.values():
            for service_model in service_models:
                target_prefix = service_model.metadata.get("targetPrefix")
                if target_prefix:
                    result[target_prefix].append(
                        ServiceModelIdentifier(service_model.service_name, service_model.protocol)
                    )
        return dict(result)

    @cached_property
    def signing_name_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        result = defaultdict(list)
        for service_models in self._services.values():
            for service_model in service_models:
                result[service_model.signing_name].append(
                    ServiceModelIdentifier(service_model.service_name, service_model.protocol)
                )
        return dict(result)

    @cached_property
    def operations_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        result = defaultdict(list)
        for service_models in self._services.values():
            for service_model in service_models:
                operations = service_model.operation_names
                if operations:
                    for operation in operations:
                        result[operation].append(
                            ServiceModelIdentifier(
                                service_model.service_name, service_model.protocol
                            )
                        )
        return dict(result)

    @cached_property
    def endpoint_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        result = defaultdict(list)
        for service_models in self._services.values():
            for service_model in service_models:
                result[service_model.endpoint_prefix].append(
                    ServiceModelIdentifier(service_model.service_name, service_model.protocol)
                )
        return dict(result)

    @cached_property
    def _services(self) -> dict[ServiceName, list[ServiceModel]]:
        services = defaultdict(list)
        for service in list_services():
            services[service.service_name].append(service)
        return services


class ServiceCatalog:
    index: ServiceCatalogIndex

    def __init__(self, index: ServiceCatalogIndex = None):
        self.index = index or LazyServiceCatalogIndex()

    @lru_cache(maxsize=512)
    def get(self, name: ServiceName, protocol: ProtocolName | None = None) -> ServiceModel | None:
        return load_service(name, protocol=protocol)

    @property
    def service_names(self) -> list[ServiceName]:
        return self.index.service_names

    @property
    def target_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        return self.index.target_prefix_index

    @property
    def signing_name_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        return self.index.signing_name_index

    @property
    def operations_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        return self.index.operations_index

    @property
    def endpoint_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:
        return self.index.endpoint_prefix_index

    def by_target_prefix(self, target_prefix: str) -> list[ServiceModelIdentifier]:
        return self.target_prefix_index.get(target_prefix, [])

    def by_signing_name(self, signing_name: str) -> list[ServiceModelIdentifier]:
        return self.signing_name_index.get(signing_name, [])

    def by_operation(self, operation_name: str) -> list[ServiceModelIdentifier]:
        return self.operations_index.get(operation_name, [])


def build_service_index_cache(file_path: str) -> ServiceCatalogIndex:
    """
    Creates a new ServiceCatalogIndex and stores it into the given file_path.

    :param file_path: the path to store the file to
    :return: the created ServiceCatalogIndex
    """
    return save_service_index_cache(LazyServiceCatalogIndex(), file_path)


def load_service_index_cache(file: str) -> ServiceCatalogIndex:
    """
    Loads from the given file the stored ServiceCatalogIndex.

    :param file: the file to load from
    :return: the loaded ServiceCatalogIndex
    """
    import dill

    with open(file, "rb") as fd:
        return dill.load(fd)


def save_service_index_cache(index: LazyServiceCatalogIndex, file_path: str) -> ServiceCatalogIndex:
    """
    Creates from the given LazyServiceCatalogIndex a ``ServiceCatalogIndex`, stores its contents into the given file,
    and then returns the newly created index.

    :param index: the LazyServiceCatalogIndex to store the index from.
    :param file_path: the path to store the binary index cache file to
    :return: the created ServiceCatalogIndex
    """
    import dill

    cache = ServiceCatalogIndex(
        service_names=index.service_names,
        endpoint_prefix_index=index.endpoint_prefix_index,
        operations_index=index.operations_index,
        signing_name_index=index.signing_name_index,
        target_prefix_index=index.target_prefix_index,
    )
    with open(file_path, "wb") as fd:
        # use dill (instead of plain pickle) to avoid issues when serializing the pickle from __main__
        dill.dump(cache, fd)
    return cache


def _get_catalog_filename():
    ls_ver = VERSION.replace(".", "_")
    botocore_ver = botocore.__version__.replace(".", "_")
    return f"service-catalog-{ls_ver}-{botocore_ver}.dill"


@singleton_factory
def get_service_catalog() -> ServiceCatalog:
    """Loads the ServiceCatalog (which contains all the service specs), and potentially re-uses a cached index."""

    try:
        catalog_file_name = _get_catalog_filename()
        static_catalog_file = os.path.join(config.dirs.static_libs, catalog_file_name)

        # try to load or load/build/save the service catalog index from the static libs
        index = None
        if os.path.exists(static_catalog_file):
            # load the service catalog from the static libs dir / built at build time
            LOG.debug("loading service catalog index cache file %s", static_catalog_file)
            index = load_service_index_cache(static_catalog_file)
        elif os.path.isdir(config.dirs.cache):
            cache_catalog_file = os.path.join(config.dirs.cache, catalog_file_name)
            if os.path.exists(cache_catalog_file):
                LOG.debug("loading service catalog index cache file %s", cache_catalog_file)
                index = load_service_index_cache(cache_catalog_file)
            else:
                LOG.debug("building service catalog index cache file %s", cache_catalog_file)
                index = build_service_index_cache(cache_catalog_file)
        return ServiceCatalog(index)
    except Exception:
        LOG.exception(
            "error while processing service catalog index cache, falling back to lazy-loaded index"
        )
        return ServiceCatalog()


def main():
    catalog_file_name = _get_catalog_filename()
    static_catalog_file = os.path.join(config.dirs.static_libs, catalog_file_name)

    if os.path.exists(static_catalog_file):
        LOG.error(
            "service catalog index cache file (%s) already there. aborting!", static_catalog_file
        )
        return 1

    # load the service catalog from the static libs dir / built at build time
    LOG.debug("building service catalog index cache file %s", static_catalog_file)
    build_service_index_cache(static_catalog_file)


if __name__ == "__main__":
    sys.exit(main())

1	import dataclasses	1✔
2	import json	1✔
3	import logging	1✔
4	import os	1✔
5	import sys	1✔
6	from collections import defaultdict	1✔
7	from collections.abc import Generator	1✔
8	from functools import cached_property, lru_cache	1✔
9	from typing import Literal, NamedTuple	1✔
10
11	import botocore	1✔
12	import jsonpatch	1✔
13	from botocore.exceptions import UnknownServiceError	1✔
14	from botocore.loaders import Loader, instance_cache	1✔
15	from botocore.model import OperationModel, ServiceModel	1✔
16
17	from localstack import config	1✔
18	from localstack.constants import VERSION	1✔
19	from localstack.utils.objects import singleton_factory	1✔
20
21	LOG = logging.getLogger(__name__)	1✔
22
23	ServiceName = str	1✔
24	ProtocolName = Literal["query", "json", "rest-json", "rest-xml", "ec2"]	1✔
25
26
27	class ServiceModelIdentifier(NamedTuple):	1✔
28	"""
29	Identifies a specific service model.
30	If the protocol is not given, the default protocol of the service with the specific name is assumed.
31	Maybe also add versions here in the future (if we can support multiple different versions for one service).
32	"""
33
34	name: ServiceName	1✔
35	protocol: ProtocolName \| None = None	1✔
36
37
38	spec_patches_json = os.path.join(os.path.dirname(__file__), "spec-patches.json")	1✔
39
40
41	def load_spec_patches() -> dict[str, list]:	1✔
42	if not os.path.exists(spec_patches_json):	1✔
UNCOV 43	return {}	×
44	with open(spec_patches_json) as fd:	1✔
45	return json.load(fd)	1✔
46
47
48	# Path for custom specs which are not (anymore) provided by botocore
49	LOCALSTACK_BUILTIN_DATA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")	1✔
50
51
52	class LocalStackBuiltInDataLoaderMixin(Loader):	1✔
53	def __init__(self, args, *kwargs):	1✔
54	# add the builtin data path to the extra_search_paths to ensure they are discovered by the loader
55	super().__init__(args, extra_search_paths=[LOCALSTACK_BUILTIN_DATA_PATH], *kwargs)	1✔
56
57
58	class PatchingLoader(Loader):	1✔
59	"""
60	A custom botocore Loader that applies JSON patches from the given json patch file to the specs as they are loaded.
61	"""
62
63	patches: dict[str, list]	1✔
64
65	def __init__(self, patches: dict[str, list], args, *kwargs):	1✔
66	# add the builtin data path to the extra_search_paths to ensure they are discovered by the loader
67	super().__init__(args, *kwargs)	1✔
68	self.patches = patches	1✔
69
70	@instance_cache	1✔
71	def load_data(self, name: str):	1✔
72	result = super(PatchingLoader, self).load_data(name)	1✔
73
74	if patches := self.patches.get(name):	1✔
75	return jsonpatch.apply_patch(result, patches)	1✔
76
77	return result	1✔
78
79
80	class CustomLoader(PatchingLoader, LocalStackBuiltInDataLoaderMixin):	1✔
81	# Class mixing the different loader features (patching, localstack specific data)
82	pass	1✔
83
84
85	loader = CustomLoader(load_spec_patches())	1✔
86
87
88	class UnknownServiceProtocolError(UnknownServiceError):	1✔
89	"""Raised when trying to load a service with an unknown protocol.
90
91	:ivar service_name: The name of the service.
92	:ivar protocol: The name of the unknown protocol.
93	"""
94
95	fmt = "Unknown service protocol: '{service_name}-{protocol}'."	1✔
96
97
98	def list_services() -> list[ServiceModel]:	1✔
99	return [load_service(service) for service in loader.list_available_services("service-2")]	1✔
100
101
102	def load_service(	1✔
103	service: ServiceName, version: str \| None = None, protocol: ProtocolName \| None = None
104	) -> ServiceModel:
105	"""
106	Loads a service
107	:param service: to load, f.e. "sqs". For custom, internalized, service protocol specs (f.e. sqs-query) it's also
108	possible to directly define the protocol in the service name (f.e. use sqs-query)
109	:param version: of the service to load, f.e. "2012-11-05", by default the latest version will be used
110	:param protocol: specific protocol to load for the specific service, f.e. "json" for the "sqs" service
111	if the service cannot be found
112	:return: Loaded service model of the service
113	:raises: UnknownServiceError if the service cannot be found
114	:raises: UnknownServiceProtocolError if the specific protocol of the service cannot be found
115	"""
116	service_description = loader.load_service_model(service, "service-2", version)	1✔
117
118	# check if the protocol is defined, and if so, if the loaded service defines this protocol
119	if protocol is not None and protocol != service_description.get("metadata", {}).get("protocol"):	1✔
120	# if the protocol is defined, but not the one of the currently loaded service,
121	# check if we already loaded the custom spec based on the naming convention (<service>-<protocol>),
122	# f.e. "sqs-query"
123	if service.endswith(f"-{protocol}"):	1✔
124	# if so, we raise an exception
UNCOV 125	raise UnknownServiceProtocolError(service_name=service, protocol=protocol)	×
126	# otherwise we try to load it (recursively)
127	try:	1✔
128	return load_service(f"{service}-{protocol}", version, protocol=protocol)	1✔
129	except UnknownServiceError:	1✔
130	# raise an unknown protocol error in case the service also can't be loaded with the naming convention
131	raise UnknownServiceProtocolError(service_name=service, protocol=protocol)	1✔
132
133	# remove potential protocol names from the service name
134	# FIXME add more protocols here if we have to internalize more than just sqs-query
135	# TODO this should not contain specific internalized serivce names
136	service = {"sqs-query": "sqs"}.get(service, service)	1✔
137	return ServiceModel(service_description, service)	1✔
138
139
140	def iterate_service_operations() -> Generator[tuple[ServiceModel, OperationModel], None, None]:	1✔
141	"""
142	Returns one record per operation in the AWS service spec, where the first item is the service model the operation
143	belongs to, and the second is the operation model.
144
145	:return: an iterable
146	"""
147	for service in list_services():	1✔
148	for op_name in service.operation_names:	1✔
149	yield service, service.operation_model(op_name)	1✔
150
151
152	@dataclasses.dataclass	1✔
153	class ServiceCatalogIndex:	1✔
154	"""
155	The ServiceCatalogIndex enables fast lookups for common operations to determine a service from service indicators.
156	"""
157
158	service_names: list[ServiceName]	1✔
159	target_prefix_index: dict[str, list[ServiceModelIdentifier]]	1✔
160	signing_name_index: dict[str, list[ServiceModelIdentifier]]	1✔
161	operations_index: dict[str, list[ServiceModelIdentifier]]	1✔
162	endpoint_prefix_index: dict[str, list[ServiceModelIdentifier]]	1✔
163
164
165	class LazyServiceCatalogIndex:	1✔
166	"""
167	A ServiceCatalogIndex that builds indexes in-memory from the spec.
168	"""
169
170	@cached_property	1✔
171	def service_names(self) -> list[ServiceName]:	1✔
172	return list(self._services.keys())	1✔
173
174	@cached_property	1✔
175	def target_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
176	result = defaultdict(list)	1✔
177	for service_models in self._services.values():	1✔
178	for service_model in service_models:	1✔
179	target_prefix = service_model.metadata.get("targetPrefix")	1✔
180	if target_prefix:	1✔
181	result[target_prefix].append(	1✔
182	ServiceModelIdentifier(service_model.service_name, service_model.protocol)
183	)
184	return dict(result)	1✔
185
186	@cached_property	1✔
187	def signing_name_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
188	result = defaultdict(list)	1✔
189	for service_models in self._services.values():	1✔
190	for service_model in service_models:	1✔
191	result[service_model.signing_name].append(	1✔
192	ServiceModelIdentifier(service_model.service_name, service_model.protocol)
193	)
194	return dict(result)	1✔
195
196	@cached_property	1✔
197	def operations_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
198	result = defaultdict(list)	1✔
199	for service_models in self._services.values():	1✔
200	for service_model in service_models:	1✔
201	operations = service_model.operation_names	1✔
202	if operations:	1✔
203	for operation in operations:	1✔
204	result[operation].append(	1✔
205	ServiceModelIdentifier(
206	service_model.service_name, service_model.protocol
207	)
208	)
209	return dict(result)	1✔
210
211	@cached_property	1✔
212	def endpoint_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
213	result = defaultdict(list)	1✔
214	for service_models in self._services.values():	1✔
215	for service_model in service_models:	1✔
216	result[service_model.endpoint_prefix].append(	1✔
217	ServiceModelIdentifier(service_model.service_name, service_model.protocol)
218	)
219	return dict(result)	1✔
220
221	@cached_property	1✔
222	def _services(self) -> dict[ServiceName, list[ServiceModel]]:	1✔
223	services = defaultdict(list)	1✔
224	for service in list_services():	1✔
225	services[service.service_name].append(service)	1✔
226	return services	1✔
227
228
229	class ServiceCatalog:	1✔
230	index: ServiceCatalogIndex	1✔
231
232	def __init__(self, index: ServiceCatalogIndex = None):	1✔
233	self.index = index or LazyServiceCatalogIndex()	1✔
234
235	@lru_cache(maxsize=512)	1✔
236	def get(self, name: ServiceName, protocol: ProtocolName \| None = None) -> ServiceModel \| None:	1✔
237	return load_service(name, protocol=protocol)	1✔
238
239	@property	1✔
240	def service_names(self) -> list[ServiceName]:	1✔
241	return self.index.service_names	1✔
242
243	@property	1✔
244	def target_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
245	return self.index.target_prefix_index	1✔
246
247	@property	1✔
248	def signing_name_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
249	return self.index.signing_name_index	1✔
250
251	@property	1✔
252	def operations_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
253	return self.index.operations_index	1✔
254
255	@property	1✔
256	def endpoint_prefix_index(self) -> dict[str, list[ServiceModelIdentifier]]:	1✔
257	return self.index.endpoint_prefix_index	1✔
258
259	def by_target_prefix(self, target_prefix: str) -> list[ServiceModelIdentifier]:	1✔
260	return self.target_prefix_index.get(target_prefix, [])	1✔
261
262	def by_signing_name(self, signing_name: str) -> list[ServiceModelIdentifier]:	1✔
263	return self.signing_name_index.get(signing_name, [])	1✔
264
265	def by_operation(self, operation_name: str) -> list[ServiceModelIdentifier]:	1✔
266	return self.operations_index.get(operation_name, [])	1✔
267
268
269	def build_service_index_cache(file_path: str) -> ServiceCatalogIndex:	1✔
270	"""
271	Creates a new ServiceCatalogIndex and stores it into the given file_path.
272
273	:param file_path: the path to store the file to
274	:return: the created ServiceCatalogIndex
275	"""
276	return save_service_index_cache(LazyServiceCatalogIndex(), file_path)	1✔
277
278
279	def load_service_index_cache(file: str) -> ServiceCatalogIndex:	1✔
280	"""
281	Loads from the given file the stored ServiceCatalogIndex.
282
283	:param file: the file to load from
284	:return: the loaded ServiceCatalogIndex
285	"""
286	import dill	1✔
287
288	with open(file, "rb") as fd:	1✔
289	return dill.load(fd)	1✔
290
291
292	def save_service_index_cache(index: LazyServiceCatalogIndex, file_path: str) -> ServiceCatalogIndex:	1✔
293	"""
294	Creates from the given LazyServiceCatalogIndex a ``ServiceCatalogIndex`, stores its contents into the given file,
295	and then returns the newly created index.
296
297	:param index: the LazyServiceCatalogIndex to store the index from.
298	:param file_path: the path to store the binary index cache file to
299	:return: the created ServiceCatalogIndex
300	"""
301	import dill	1✔
302
303	cache = ServiceCatalogIndex(	1✔
304	service_names=index.service_names,
305	endpoint_prefix_index=index.endpoint_prefix_index,
306	operations_index=index.operations_index,
307	signing_name_index=index.signing_name_index,
308	target_prefix_index=index.target_prefix_index,
309	)
310	with open(file_path, "wb") as fd:	1✔
311	# use dill (instead of plain pickle) to avoid issues when serializing the pickle from __main__
312	dill.dump(cache, fd)	1✔
313	return cache	1✔
314
315
316	def _get_catalog_filename():	1✔
317	ls_ver = VERSION.replace(".", "_")	1✔
318	botocore_ver = botocore.__version__.replace(".", "_")	1✔
319	return f"service-catalog-{ls_ver}-{botocore_ver}.dill"	1✔
320
321
322	@singleton_factory	1✔
323	def get_service_catalog() -> ServiceCatalog:	1✔
324	"""Loads the ServiceCatalog (which contains all the service specs), and potentially re-uses a cached index."""
325
326	try:	1✔
327	catalog_file_name = _get_catalog_filename()	1✔
328	static_catalog_file = os.path.join(config.dirs.static_libs, catalog_file_name)	1✔
329
330	# try to load or load/build/save the service catalog index from the static libs
331	index = None	1✔
332	if os.path.exists(static_catalog_file):	1✔
333	# load the service catalog from the static libs dir / built at build time
334	LOG.debug("loading service catalog index cache file %s", static_catalog_file)	1✔
335	index = load_service_index_cache(static_catalog_file)	1✔
336	elif os.path.isdir(config.dirs.cache):	1✔
337	cache_catalog_file = os.path.join(config.dirs.cache, catalog_file_name)	1✔
338	if os.path.exists(cache_catalog_file):	1✔
UNCOV 339	LOG.debug("loading service catalog index cache file %s", cache_catalog_file)	×
340	index = load_service_index_cache(cache_catalog_file)	×
341	else:
342	LOG.debug("building service catalog index cache file %s", cache_catalog_file)	1✔
343	index = build_service_index_cache(cache_catalog_file)	1✔
344	return ServiceCatalog(index)	1✔
UNCOV 345	except Exception:	×
346	LOG.exception(	×
347	"error while processing service catalog index cache, falling back to lazy-loaded index"
348	)
UNCOV 349	return ServiceCatalog()	×
350
351
352	def main():	1✔
UNCOV 353	catalog_file_name = _get_catalog_filename()	×
354	static_catalog_file = os.path.join(config.dirs.static_libs, catalog_file_name)	×
355
UNCOV 356	if os.path.exists(static_catalog_file):	×
357	LOG.error(	×
358	"service catalog index cache file (%s) already there. aborting!", static_catalog_file
359	)
UNCOV 360	return 1	×
361
362	# load the service catalog from the static libs dir / built at build time
UNCOV 363	LOG.debug("building service catalog index cache file %s", static_catalog_file)	×
364	build_service_index_cache(static_catalog_file)	×
365
366
367	if __name__ == "__main__":
368	sys.exit(main())

localstack / localstack / 16820655284

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