How to use invocation_response method in localstack

Best Python code snippet using localstack_python

base.py

Source:base.py

1# -*- coding: utf-8 -*-2"""3Copyright (C) 2021, Zato Source s.r.o. https://zato.io4Licensed under LGPLv3, see LICENSE.txt for terms and conditions.5"""6# stdlib7from datetime import datetime8from logging import getLogger9# Zato10from zato.common import CHANNEL11from zato.common.util import spawn_greenlet12from zato.server.pattern.model import CacheEntry, InvocationResponse, ParallelCtx, Target13# ################################################################################################################################14if 0:15    from gevent.lock import RLock16    from zato.server.service import Service17    Service = Service18# ################################################################################################################################19logger = getLogger(__name__)20# ################################################################################################################################21# ################################################################################################################################22class ParallelBase:23    """ A base class for most parallel integration patterns. An instance of this class is created for each service instance.24    """25    call_channel = '<parallel-base-call-channel-not-set>'26    on_target_channel = '<parallel-base-target-channel-not-set>'27    on_final_channel = '<parallel-base-final-channel-not-set>'28    needs_on_final = False29    def __init__(self, source, cache, lock):30        # type: (Service, dict, RLock) -> None31        self.source = source32        self.cache = cache33        self.lock = lock34        self.cid = source.cid35# ################################################################################################################################36    def _invoke(self, ctx):37        # type: (ParallelCtx)38        # Store metadata about our invocation ..39        with self.lock:40            # .. create a new entry ..41            entry = CacheEntry()42            entry.cid = ctx.cid43            entry.req_ts_utc = ctx.req_ts_utc44            entry.len_targets = len(ctx.target_list)45            entry.remaining_targets = entry.len_targets46            entry.target_responses = []47            entry.final_responses = {}48            entry.on_target_list = ctx.on_target_list49            entry.on_final_list = ctx.on_final_list50            # .. and add it to the cache.51            self.cache[ctx.cid] = entry52        # Now that metadata is stored, we can actually invoke each of the serviced from our list of targets.53        for item in ctx.target_list: # type: Target54            self.source.invoke_async(item.name, item.payload, channel=self.call_channel, cid=ctx.cid)55# ################################################################################################################################56    def invoke(self, targets, on_final, on_target=None, cid=None, _utcnow=datetime.utcnow):57        """ Invokes targets collecting their responses, can be both as a whole or individual ones,58        and executes callback(s).59        """60        # type: (dict, list, list, str, object) -> None61        # Establish what our CID is ..62        cid = cid or self.cid63        # .. set up targets to invoke ..64        target_list = []65        for target_name, payload in targets.items():66            target = Target()67            target.name = target_name68            target.payload = payload69            target_list.append(target)70        # .. create an execution context ..71        ctx = ParallelCtx()72        ctx.cid = cid73        ctx.req_ts_utc = _utcnow()74        ctx.source_name = self.source.name75        ctx.target_list = target_list76        # .. on-final is always available ..77        ctx.on_final_list = [on_final] if isinstance(on_final, str) else on_final78        # .. but on-target may be None ..79        if on_target:80            ctx.on_target_list = [on_target] if isinstance(on_target, str) else on_target81        # .. invoke our implementation in background ..82        try:83            spawn_greenlet(self._invoke, ctx)84        except Exception:85            # Explicitly ignore any exception caught - this is because we are catching86            # deeper in the call stack to provide it to callback services so we do not want87            # to raise it here too.88            pass89        # .. and return the CID to the caller.90        return cid91# ################################################################################################################################92    def on_call_finished(self, invoked_service, response, exception, _utcnow=datetime.utcnow):93        # type: (Service, object, Exception, object)94        # Update metadata about the current parallel execution under a server-wide lock ..95        with self.lock:96            # .. find our cache entry ..97            entry = self.cache.get(invoked_service.cid) # type: CacheEntry98            # .. exit early if we cannot find the entry for any reason ..99            if not entry:100                logger.warning('No such parallel cache key `%s`', invoked_service.cid)101                return102            # .. alright, we can proceed ..103            else:104                # .. update the number of targets already invoked ..105                entry.remaining_targets -= 1106                # .. build information about the response that we have ..107                invocation_response = InvocationResponse()108                invocation_response.cid = invoked_service.cid109                invocation_response.req_ts_utc = entry.req_ts_utc110                invocation_response.resp_ts_utc = _utcnow()111                invocation_response.response = response112                invocation_response.exception = exception113                invocation_response.ok = False if exception else True114                invocation_response.source = self.source.name115                invocation_response.target = invoked_service.name116                # For pre-Zato 3.2 compatibility, callbacks expect dicts on input.117                dict_payload = {118                    'source': invocation_response.source,119                    'target': invocation_response.target,120                    'response': invocation_response.response,121                    'req_ts_utc': invocation_response.req_ts_utc.isoformat(),122                    'resp_ts_utc': invocation_response.resp_ts_utc.isoformat(),123                    'ok': invocation_response.ok,124                    'exception': invocation_response.exception,125                    'cid': invocation_response.cid,126                }127                # .. add the received response to the list of what we have so far ..128                entry.target_responses.append(dict_payload)129                # .. invoke any potential on-target callbacks ..130                if entry.on_target_list:131                    # Updates the dictionary in-place132                    dict_payload['phase'] = 'on-target'133                    for on_target_item in entry.on_target_list: # type: str134                        invoked_service.invoke_async(135                            on_target_item, dict_payload, channel=self.on_target_channel, cid=invoked_service.cid)136                # .. check if this was the last service that we were waiting for ..137                if entry.remaining_targets == 0:138                    # .. if so, run the final callback services if it is required in our case ..139                    if self.needs_on_final:140                        if entry.on_final_list:141                            # This message is what all the on-final callbacks142                            # receive in their self.request.payload attribute.143                            on_final_message = {144                                'phase': 'on-final',145                                'source': invocation_response.source,146                                'req_ts_utc': entry.req_ts_utc,147                                'on_target': entry.on_target_list,148                                'on_final': entry.on_final_list,149                                'data': entry.target_responses,150                            }151                            for on_final_item in entry.on_final_list: # type: str152                                invoked_service.invoke_async(153                                    on_final_item, on_final_message,154                                    channel=self.on_final_channel, cid=invoked_service.cid)155                    # .. now, clean up by deleting the current entry from cache.156                    # Note that we ise None in an unlikely it is already deleted,157                    # although this should not happen because we are the only piece of code holding this lock.158                    self.cache.pop(invoked_service.cid, None)159# ################################################################################################################################160# ################################################################################################################################161class ParallelExec(ParallelBase):162    call_channel = CHANNEL.PARALLEL_EXEC_CALL163    on_target_channel = CHANNEL.PARALLEL_EXEC_ON_TARGET164    def invoke(self, targets, on_target, cid=None):165        return super().invoke(targets, None, on_target, cid)166# ################################################################################################################################167# ################################################################################################################################168class FanOut(ParallelBase):169    call_channel = CHANNEL.FANOUT_CALL170    on_target_channel = CHANNEL.FANOUT_ON_TARGET171    on_final_channel = CHANNEL.FANOUT_ON_FINAL172    needs_on_final = True173# ################################################################################################################################...

bootstrap.py

Source:bootstrap.py

1ort os2import requests3import sys4import traceback5def run_loop():6       aws_lambda_runtime_api = os.environ['AWS_LAMBDA_RUNTIME_API']7       8       import app9       10       while True:11              request_id = None12              try:13                     invocation_response = requests.get(f'http://{aws_lambda_runtime_api}/2018-06-01/runtime/invocation/next')14                     request_id = invocation_response.headers['Lambda-Runtime-Aws-Request-Id']15                     invoked_function_arn = invocation_response.headers['Lambda-Runtime-Invoked-Function-Arn']16                     trace_id = invocation_response.headers['Lambda-Runtime-Trace-Id']17                     os.environ['_X_AMZN_TRACE_ID'] = trace_id18                     19                     context = {20                            'request_id': request_id,21                            'invoked_function_arn': invoked_function_arn,22                            'trace_id': trace_id23                     }24                     25                     event = invocation_response.json()26                     27                     response_url = f'http://{aws_lambda_runtime_api}/2018-06-01/runtime/invocation/{request_id}/response'28                     29                     result = app.lambda_handler(event, context)30                     31                     sys.stdout.flush()32                     requests.post(response_url, json=result)33              34              except:35                      if request_id != None:36                            try:37                                   exc_type, exc_value, exc_traceback = sys.exc_info()38                                   exception_message = {39                                        'errorType': exc_type.__name__,40                                        'errorMessage': str(exc_value),41                                        'stackTrace': traceback.format_exception(exc_type,exc_value, exc_traceback)42                                   }43                                   44                                   error_url = f'http://{aws_lambda_runtime_api}/2018-06-01/runtime/invocation/{request_id}/error'45                                   sys.stdout.flush()46                                   47                                   requests.post(error_url, json=exception_message)48                            except:49                                   pass...

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.