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

95.37

/localstack-core/localstack/aws/skeleton.py

import inspect
import logging
from collections.abc import Callable
from typing import Any, NamedTuple

from botocore import xform_name
from botocore.model import ServiceModel

from localstack.aws.api import (
    CommonServiceException,
    RequestContext,
    ServiceException,
)
from localstack.aws.api.core import ServiceRequest, ServiceRequestHandler, ServiceResponse
from localstack.aws.protocol.parser import create_parser
from localstack.aws.protocol.serializer import ResponseSerializer, create_serializer
from localstack.aws.spec import load_service
from localstack.http import Response
from localstack.utils import analytics
from localstack.utils.coverage_docs import get_coverage_link_for_service

LOG = logging.getLogger(__name__)

DispatchTable = dict[str, ServiceRequestHandler]


def create_skeleton(service: str | ServiceModel, delegate: Any):
    if isinstance(service, str):
        service = load_service(service)

    return Skeleton(service, create_dispatch_table(delegate))


class HandlerAttributes(NamedTuple):
    """
    Holder object of the attributes added to a function by the @handler decorator.
    """

    function_name: str
    operation: str
    pass_context: bool
    expand_parameters: bool


def create_dispatch_table(delegate: object) -> DispatchTable:
    """
    Creates a dispatch table for a given object. First, the entire class tree of the object is scanned to find any
    functions that are decorated with @handler. It then resolves those functions on the delegate.
    """
    # scan class tree for @handler wrapped functions (reverse class tree so that inherited functions overwrite parent
    # functions)
    cls_tree = inspect.getmro(delegate.__class__)
    handlers: dict[str, HandlerAttributes] = {}
    cls_tree = reversed(list(cls_tree))
    for cls in cls_tree:
        if cls == object:
            continue

        for name, fn in inspect.getmembers(cls, inspect.isfunction):
            try:
                # attributes come from operation_marker in @handler wrapper
                handlers[fn.operation] = HandlerAttributes(
                    fn.__name__, fn.operation, fn.pass_context, fn.expand_parameters
                )
            except AttributeError:
                pass

    # create dispatch table from operation handlers by resolving bound functions on the delegate
    dispatch_table: DispatchTable = {}
    for handler in handlers.values():
        # resolve the bound function of the delegate
        bound_function = getattr(delegate, handler.function_name)
        # create a dispatcher
        dispatch_table[handler.operation] = ServiceRequestDispatcher(
            bound_function,
            operation=handler.operation,
            pass_context=handler.pass_context,
            expand_parameters=handler.expand_parameters,
        )

    return dispatch_table


class ServiceRequestDispatcher:
    fn: Callable
    operation: str
    expand_parameters: bool = True
    pass_context: bool = True

    def __init__(
        self,
        fn: Callable,
        operation: str,
        pass_context: bool = True,
        expand_parameters: bool = True,
    ):
        self.fn = fn
        self.operation = operation
        self.pass_context = pass_context
        self.expand_parameters = expand_parameters

    def __call__(self, context: RequestContext, request: ServiceRequest) -> ServiceResponse | None:
        args = []
        kwargs = {}

        if not self.expand_parameters:
            if self.pass_context:
                args.append(context)
            args.append(request)
        else:
            if request is None:
                kwargs = {}
            else:
                kwargs = {xform_name(k): v for k, v in request.items()}
            kwargs["context"] = context

        return self.fn(*args, **kwargs)


class Skeleton:
    service: ServiceModel
    dispatch_table: DispatchTable

    def __init__(self, service: ServiceModel, implementation: Any | DispatchTable):
        self.service = service

        if isinstance(implementation, dict):
            self.dispatch_table = implementation
        else:
            self.dispatch_table = create_dispatch_table(implementation)

    def invoke(self, context: RequestContext) -> Response:
        serializer = create_serializer(context.service)

        if context.operation and context.service_request:
            # if the parsed request is already set in the context, re-use them
            operation, instance = context.operation, context.service_request
        else:
            # otherwise, parse the incoming HTTPRequest
            operation, instance = create_parser(context.service).parse(context.request)
            context.operation = operation

        try:
            # Find the operation's handler in the dispatch table
            if operation.name not in self.dispatch_table:
                LOG.warning(
                    "missing entry in dispatch table for %s.%s",
                    self.service.service_name,
                    operation.name,
                )
                raise NotImplementedError

            return self.dispatch_request(serializer, context, instance)
        except ServiceException as e:
            return self.on_service_exception(serializer, context, e)
        except NotImplementedError as e:
            return self.on_not_implemented_error(serializer, context, e)

    def dispatch_request(
        self, serializer: ResponseSerializer, context: RequestContext, instance: ServiceRequest
    ) -> Response:
        operation = context.operation

        handler = self.dispatch_table[operation.name]

        # Call the appropriate handler
        result = handler(context, instance) or {}

        # if the service handler returned an HTTP request, forego serialization and return immediately
        if isinstance(result, Response):
            return result

        context.service_response = result

        # Serialize result dict to a Response and return it
        return serializer.serialize_to_response(
            result, operation, context.request.headers, context.request_id
        )

    def on_service_exception(
        self, serializer: ResponseSerializer, context: RequestContext, exception: ServiceException
    ) -> Response:
        """
        Called by invoke if the handler of the operation raised a ServiceException.

        :param serializer: serializer which should be used to serialize the exception
        :param context: the request context
        :param exception: the exception that was raised
        :return: a Response object
        """
        context.service_exception = exception

        return serializer.serialize_error_to_response(
            exception, context.operation, context.request.headers, context.request_id
        )

    def on_not_implemented_error(
        self,
        serializer: ResponseSerializer,
        context: RequestContext,
        exception: NotImplementedError,
    ) -> Response:
        """
        Called by invoke if either the dispatch table did not contain an entry for the operation, or the service
        provider raised a NotImplementedError
        :param serializer: the serialzier which should be used to serialize the NotImplementedError
        :param context: the request context
        :param exception: the NotImplementedError that was raised
        :return: a Response object
        """
        operation = context.operation

        action_name = operation.name
        service_name = operation.service_model.service_name
        exception_message: str | None = exception.args[0] if exception.args else None
        message = exception_message or get_coverage_link_for_service(service_name, action_name)
        LOG.info(message)
        error = CommonServiceException("InternalFailure", message, status_code=501)
        # record event
        analytics.log.event(
            "services_notimplemented", payload={"s": service_name, "a": action_name}
        )
        context.service_exception = error

        return serializer.serialize_error_to_response(
            error, operation, context.request.headers, context.request_id
        )

1	import inspect	1✔
2	import logging	1✔
3	from collections.abc import Callable	1✔
4	from typing import Any, NamedTuple	1✔
5
6	from botocore import xform_name	1✔
7	from botocore.model import ServiceModel	1✔
8
9	from localstack.aws.api import (	1✔
10	CommonServiceException,
11	RequestContext,
12	ServiceException,
13	)
14	from localstack.aws.api.core import ServiceRequest, ServiceRequestHandler, ServiceResponse	1✔
15	from localstack.aws.protocol.parser import create_parser	1✔
16	from localstack.aws.protocol.serializer import ResponseSerializer, create_serializer	1✔
17	from localstack.aws.spec import load_service	1✔
18	from localstack.http import Response	1✔
19	from localstack.utils import analytics	1✔
20	from localstack.utils.coverage_docs import get_coverage_link_for_service	1✔
21
22	LOG = logging.getLogger(__name__)	1✔
23
24	DispatchTable = dict[str, ServiceRequestHandler]	1✔
25
26
27	def create_skeleton(service: str \| ServiceModel, delegate: Any):	1✔
28	if isinstance(service, str):	×
UNCOV 29	service = load_service(service)	×
30
UNCOV 31	return Skeleton(service, create_dispatch_table(delegate))	×
32
33
34	class HandlerAttributes(NamedTuple):	1✔
35	"""
36	Holder object of the attributes added to a function by the @handler decorator.
37	"""
38
39	function_name: str	1✔
40	operation: str	1✔
41	pass_context: bool	1✔
42	expand_parameters: bool	1✔
43
44
45	def create_dispatch_table(delegate: object) -> DispatchTable:	1✔
46	"""
47	Creates a dispatch table for a given object. First, the entire class tree of the object is scanned to find any
48	functions that are decorated with @handler. It then resolves those functions on the delegate.
49	"""
50	# scan class tree for @handler wrapped functions (reverse class tree so that inherited functions overwrite parent
51	# functions)
52	cls_tree = inspect.getmro(delegate.__class__)	1✔
53	handlers: dict[str, HandlerAttributes] = {}	1✔
54	cls_tree = reversed(list(cls_tree))	1✔
55	for cls in cls_tree:	1✔
56	if cls == object:	1✔
57	continue	1✔
58
59	for name, fn in inspect.getmembers(cls, inspect.isfunction):	1✔
60	try:	1✔
61	# attributes come from operation_marker in @handler wrapper
62	handlers[fn.operation] = HandlerAttributes(	1✔
63	fn.__name__, fn.operation, fn.pass_context, fn.expand_parameters
64	)
65	except AttributeError:	1✔
66	pass	1✔
67
68	# create dispatch table from operation handlers by resolving bound functions on the delegate
69	dispatch_table: DispatchTable = {}	1✔
70	for handler in handlers.values():	1✔
71	# resolve the bound function of the delegate
72	bound_function = getattr(delegate, handler.function_name)	1✔
73	# create a dispatcher
74	dispatch_table[handler.operation] = ServiceRequestDispatcher(	1✔
75	bound_function,
76	operation=handler.operation,
77	pass_context=handler.pass_context,
78	expand_parameters=handler.expand_parameters,
79	)
80
81	return dispatch_table	1✔
82
83
84	class ServiceRequestDispatcher:	1✔
85	fn: Callable	1✔
86	operation: str	1✔
87	expand_parameters: bool = True	1✔
88	pass_context: bool = True	1✔
89
90	def __init__(	1✔
91	self,
92	fn: Callable,
93	operation: str,
94	pass_context: bool = True,
95	expand_parameters: bool = True,
96	):
97	self.fn = fn	1✔
98	self.operation = operation	1✔
99	self.pass_context = pass_context	1✔
100	self.expand_parameters = expand_parameters	1✔
101
102	def __call__(self, context: RequestContext, request: ServiceRequest) -> ServiceResponse \| None:	1✔
103	args = []	1✔
104	kwargs = {}	1✔
105
106	if not self.expand_parameters:	1✔
107	if self.pass_context:	1✔
108	args.append(context)	1✔
109	args.append(request)	1✔
110	else:
111	if request is None:	1✔
UNCOV 112	kwargs = {}	×
113	else:
114	kwargs = {xform_name(k): v for k, v in request.items()}	1✔
115	kwargs["context"] = context	1✔
116
117	return self.fn(args, *kwargs)	1✔
118
119
120	class Skeleton:	1✔
121	service: ServiceModel	1✔
122	dispatch_table: DispatchTable	1✔
123
124	def __init__(self, service: ServiceModel, implementation: Any \| DispatchTable):	1✔
125	self.service = service	1✔
126
127	if isinstance(implementation, dict):	1✔
128	self.dispatch_table = implementation	1✔
129	else:
130	self.dispatch_table = create_dispatch_table(implementation)	1✔
131
132	def invoke(self, context: RequestContext) -> Response:	1✔
133	serializer = create_serializer(context.service)	1✔
134
135	if context.operation and context.service_request:	1✔
136	# if the parsed request is already set in the context, re-use them
137	operation, instance = context.operation, context.service_request	1✔
138	else:
139	# otherwise, parse the incoming HTTPRequest
140	operation, instance = create_parser(context.service).parse(context.request)	1✔
141	context.operation = operation	1✔
142
143	try:	1✔
144	# Find the operation's handler in the dispatch table
145	if operation.name not in self.dispatch_table:	1✔
146	LOG.warning(	1✔
147	"missing entry in dispatch table for %s.%s",
148	self.service.service_name,
149	operation.name,
150	)
151	raise NotImplementedError
152
153	return self.dispatch_request(serializer, context, instance)	1✔
154	except ServiceException as e:	1✔
155	return self.on_service_exception(serializer, context, e)	1✔
156	except NotImplementedError as e:	1✔
157	return self.on_not_implemented_error(serializer, context, e)	1✔
158
159	def dispatch_request(	1✔
160	self, serializer: ResponseSerializer, context: RequestContext, instance: ServiceRequest
161	) -> Response:
162	operation = context.operation	1✔
163
164	handler = self.dispatch_table[operation.name]	1✔
165
166	# Call the appropriate handler
167	result = handler(context, instance) or {}	1✔
168
169	# if the service handler returned an HTTP request, forego serialization and return immediately
170	if isinstance(result, Response):	1✔
UNCOV 171	return result	×
172
173	context.service_response = result	1✔
174
175	# Serialize result dict to a Response and return it
176	return serializer.serialize_to_response(	1✔
177	result, operation, context.request.headers, context.request_id
178	)
179
180	def on_service_exception(	1✔
181	self, serializer: ResponseSerializer, context: RequestContext, exception: ServiceException
182	) -> Response:
183	"""
184	Called by invoke if the handler of the operation raised a ServiceException.
185
186	:param serializer: serializer which should be used to serialize the exception
187	:param context: the request context
188	:param exception: the exception that was raised
189	:return: a Response object
190	"""
191	context.service_exception = exception	1✔
192
193	return serializer.serialize_error_to_response(	1✔
194	exception, context.operation, context.request.headers, context.request_id
195	)
196
197	def on_not_implemented_error(	1✔
198	self,
199	serializer: ResponseSerializer,
200	context: RequestContext,
201	exception: NotImplementedError,
202	) -> Response:
203	"""
204	Called by invoke if either the dispatch table did not contain an entry for the operation, or the service
205	provider raised a NotImplementedError
206	:param serializer: the serialzier which should be used to serialize the NotImplementedError
207	:param context: the request context
208	:param exception: the NotImplementedError that was raised
209	:return: a Response object
210	"""
211	operation = context.operation	1✔
212
213	action_name = operation.name	1✔
214	service_name = operation.service_model.service_name	1✔
215	exception_message: str \| None = exception.args[0] if exception.args else None	1✔
216	message = exception_message or get_coverage_link_for_service(service_name, action_name)	1✔
217	LOG.info(message)	1✔
218	error = CommonServiceException("InternalFailure", message, status_code=501)	1✔
219	# record event
220	analytics.log.event(	1✔
221	"services_notimplemented", payload={"s": service_name, "a": action_name}
222	)
223	context.service_exception = error	1✔
224
225	return serializer.serialize_error_to_response(	1✔
226	error, operation, context.request.headers, context.request_id
227	)

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