Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use record_parsed_request method in localstack

Best Python code snippet using localstack_python

metric_handler.py

Source:metric_handler.py

...108        item = MetricHandlerItem(context)109        self.metrics_handler_items[context] = item110    def _get_metric_handler_item_for_context(self, context: RequestContext) -> MetricHandlerItem:111        return self.metrics_handler_items[context]112    def record_parsed_request(113        self, chain: HandlerChain, context: RequestContext, response: Response114    ):115        if not config.is_collect_metrics_mode():116            return117        item = self._get_metric_handler_item_for_context(context)118        item.parameters_after_parse = (119            list(context.service_request.keys()) if context.service_request else []120        )121    def record_exception(122        self, chain: HandlerChain, exception: Exception, context: RequestContext, response: Response123    ):124        if not config.is_collect_metrics_mode():125            return126        item = self._get_metric_handler_item_for_context(context)...

app.py

Source:app.py

1from localstack.aws import handlers2from localstack.aws.handlers.metric_handler import MetricHandler3from localstack.aws.handlers.service_plugin import ServiceLoader4from localstack.services.plugins import SERVICE_PLUGINS, ServiceManager, ServicePluginManager5from .gateway import Gateway6from .handlers.fallback import EmptyResponseHandler7from .handlers.service import ServiceRequestRouter8class LocalstackAwsGateway(Gateway):9    def __init__(self, service_manager: ServiceManager = None) -> None:10        super().__init__()11        # basic server components12        self.service_manager = service_manager or ServicePluginManager()13        self.service_request_router = ServiceRequestRouter()14        # lazy-loads services into the router15        load_service = ServiceLoader(self.service_manager, self.service_request_router)16        metric_collector = MetricHandler()17        # the main request handler chain18        self.request_handlers.extend(19            [20                handlers.push_request_context,21                metric_collector.create_metric_handler_item,22                handlers.parse_service_name,  # enforce_cors and content_decoder depend on the service name23                handlers.enforce_cors,24                handlers.content_decoder,25                handlers.serve_localstack_resources,  # try to serve internal resources in /_localstack first26                handlers.serve_default_listeners,  # legacy proxy default listeners27                handlers.serve_edge_router_rules,28                # start aws handler chain29                handlers.inject_auth_header_if_missing,30                handlers.add_region_from_header,31                handlers.add_account_id,32                handlers.parse_service_request,33                metric_collector.record_parsed_request,34                handlers.serve_custom_service_request_handlers,35                load_service,  # once we have the service request we can make sure we load the service36                self.service_request_router,  # once we know the service is loaded we can route the request37                # if the chain is still running, set an empty response38                EmptyResponseHandler(404, b'{"message": "Not Found"}'),39            ]40        )41        # exception handlers in the chain42        self.exception_handlers.extend(43            [44                handlers.log_exception,45                handlers.handle_service_exception,46                handlers.handle_internal_failure,47            ]48        )49        # response post-processing50        self.response_handlers.extend(51            [52                handlers.parse_service_response,53                handlers.set_close_connection_header,54                handlers.run_custom_response_handlers,55                handlers.add_cors_response_headers,56                handlers.log_response,57                handlers.count_service_request,58                handlers.pop_request_context,59                metric_collector.update_metric_collection,60            ]61        )62def main():63    """64    Serve the LocalstackGateway with the default configuration directly through hypercorn. This is mostly for65    development purposes and documentation on how to serve the Gateway.66    """67    from .serving.hypercorn import serve68    use_ssl = True69    port = 456670    # serve the LocalStackAwsGateway in a dev app71    from localstack.utils.bootstrap import setup_logging72    setup_logging()73    if use_ssl:74        from localstack.services.generic_proxy import (75            GenericProxy,76            install_predefined_cert_if_available,77        )78        install_predefined_cert_if_available()79        _, cert_file_name, key_file_name = GenericProxy.create_ssl_cert(serial_number=port)80        ssl_creds = (cert_file_name, key_file_name)81    else:82        ssl_creds = None83    gw = LocalstackAwsGateway(SERVICE_PLUGINS)84    serve(gw, use_reloader=True, port=port, ssl_creds=ssl_creds)85if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.