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

How to use delete_gateway_response method in localstack

Best Python code snippet using localstack_python

deleteThings.py

Source:deleteThings.py

1import boto32import botocore3import json4from boto3.dynamodb.conditions import Key5GATEWAY_TYPE = "gateway"6SENSOR_TYPE = "sentimate"7isGatewayPolicy = False8policyName = "gateway-policy"9iot = boto3.client('iot')10response = {}11def deleteDevice(payload):12    global response13    outputObj = {}14    for data in payload:15        if(data["deviceType"] == GATEWAY_TYPE):16            print("Device type is: {}".format(GATEWAY_TYPE))17            #macAddress = data["macAddress"]18            #groupName = "group-" + GATEWAY_TYPE + "-" + macAddress19            groupName = data["gatewayId"]20            thingName = data["gatewayId"]21            print("Listing associated principals with thing: {}".format(thingName))22            try:23                r = iot.list_thing_principals(thingName=thingName)24            except botocore.exceptions.ClientError as error:25                print("Error listing principals with the thing: {}".format(thingName))26                print(error.response)27                print(error.response['Error']['Code'])28                response["message"] = "Error Occcured while listing thing principals"29                response["errorType"] = error.response['Error']['Code']30                response["errorCode"] = 50031                return response32            certificateArn = r['principals'][0]33            print("Detaching policy from the certificate: {}".format(certificateArn))34            try:35                iot.detach_policy(policyName=policyName, target=certificateArn)36                print("Successfully detached policy!")37            except botocore.exceptions.ClientError as error:38                print("Error detaching the policy")39                print(error.response['Error']['Code'])40                response["message"] = "Error Occcured detaching policy"41                response["errorType"] = error.response['Error']['Code']42                response["errorCode"] = 50043                return response44            print("Detaching the certificate from the thing: {}".format(thingName))45            try:46                iot.detach_thing_principal(thingName=thingName, principal=certificateArn)47                print("Successfully detached certificate!")48            except botocore.exceptions.ClientError as error:49                print("Error detaching the certificate from the thing: {}".format(thingName))50                print(error.response['Error']['Code'])51                response["message"] = "Error Occcured detaching the certificate from the thing"52                response["errorType"] = error.response['Error']['Code']53                response["errorCode"] = 50054                return response55            certificateId=certificateArn.split('/')[1]56            print("Updating the certificate: {}".format(certificateArn))  57            try:58                iot.update_certificate(certificateId=certificateId, newStatus='INACTIVE')59                print("Successfully updated certificate to inactive!")60            except botocore.exceptions.ClientError as error:61                print("Error deleting the certificate: {}".format(certificateArn))62                print(error.response['Error']['Code'])63                response["message"] = "Error Occcured updating the certificate"64                response["errorType"] = error.response['Error']['Code']65                response["errorCode"] = 50066                return response67            print("Deleting the certificate: {}".format(certificateArn))68            69            try:70                iot.delete_certificate(certificateId=certificateId, forceDelete=True)71                print("Successfully deleted certificate!")72            except botocore.exceptions.ClientError as error:73                print("Error deleting the certificate: {}".format(certificateArn))74                print(error.response['Error']['Code'])75                response["message"] = "Error Occcured deleting the certificate"76                response["errorType"] = error.response['Error']['Code']77                response["errorCode"] = 50078                return response79            print("Deleting thing: {}".format(thingName))80            try:81                r = iot.delete_thing(thingName=thingName)82                print("Successfully deleted the thing...")83            except botocore.exceptions.ClientError as error:84                print("Error deleting the thing: {}".format(thingName))85                print(error.response['Error']['Code'])86                response["message"] = "Error Occcured deleting the thing"87                response["errorType"] = error.response['Error']['Code']88                response["errorCode"] = 50089                return response90            print("Deleting the group: {}".format(groupName))91            try:92                r = iot.delete_thing_group(thingGroupName=groupName)93                print("Successfully deleted the thing group: {}".format(groupName))94                response["message"] = "Successfully deleted!"95                response["errorCode"]  = 20096            except botocore.exceptions.ClientError as error:97                print("Error deleting the thing group: {}".format(groupName))98                print(error.response['Error']['Code'])99                response["message"] = "Error Occcured deleting the group"100                response["errorType"] = error.response['Error']['Code']101                response["errorCode"] = 500102                return response103            104            gateway_id = groupName105            gateway_response=checkGateway(gateway_id)106            if 'Item' in gateway_response:107                delete_gateway_response=deleteGateway(gateway_id)108                if delete_gateway_response['ResponseMetadata']['HTTPStatusCode']==200:109                    outputObj['statusCode']=200110                    outputObj['message']="Gateway deleted successfully"111                    response["db"] = outputObj   112                else:113                    outputObj['statusCode']=500114                    outputObj['message']="Internal server error"115                    response["db"] = outputObj116            else:117                outputObj['statusCode']=200118                outputObj['message']="Gateway not found in database"119                response["db"] = outputObj120            121        elif(data["deviceType"] == SENSOR_TYPE):122            print("Device type is: {}".format(SENSOR_TYPE))123            #macAddress = data["macAddress"]124            groupName = data["gatewayId"]125            #thingName = SENSOR_TYPE + "-" + macAddress126            thingName = data["sensorId"]127            print("Deleting thing: {}".format(thingName))128            try:129                r = iot.delete_thing(thingName=thingName)130                print("Successfully deleted the thing...")131                response["message"] = "Successfully deleted!"132                response["errorCode"] = 200133            except botocore.exceptions.ClientError as error:134                print("Error deleting the thing: {}".format(thingName))135                print(error.response['Error']['Code'])136                response["message"] = "Error Occcured deleting the thing {}".format(thingName)137                response["errorType"] = error.response['Error']['Code']138                response["errorCode"] = 500139                return response140            141            142            gatewayId=groupName143            sensorId=data["sensorId"]144            gateway_response=checkGateway(gatewayId)145            if 'Item' in gateway_response:146                print("Gateway Response: {}".format(gateway_response))147                gateway=gateway_response['Item']148                print("gateway: {}".format(gateway))149                sensors=gateway['sensors']150                print("Sensors: {}".format(sensors))151                newSensors=[]152                for i in sensors:153                    print("in for loop: ")154                    print(i)155                    if i['sensorId']==sensorId:156                        print("found match")157                    else:158                        newSensors.append(i)159                dynamodb_table=getClient()160                table = dynamodb_table.Table('devices')       161                response1 = table.update_item(162                        Key={163                            'gatewayId': gatewayId164                        },165                        UpdateExpression="set #s = :r", 166                        ExpressionAttributeNames={167                        '#s': 'sensors'168                        },169                        ExpressionAttributeValues={170                            ':r': newSensors,171                        },172                        ReturnValues="UPDATED_NEW"173                        )174                print(response1)175                if 'Attributes' in response1:176                    response['errorCode'] = 200177                    response['message'] = 'Successfully deleted!'178                    outputObj['statusCode']=200179                    outputObj['message']="Sensors deleted successfully in dynamodb"180                    response["db"] = outputObj181        182    return response183    184def checkGateway(gatewayId):185    dynamodb_table=getClient()186    table = dynamodb_table.Table('devices')187    response = table.get_item(188        Key={189        'gatewayId':gatewayId190        }191        )192    return response193    194def deleteGateway(gatewayId):195    dynamodb_table=getClient()196    table = dynamodb_table.Table('devices')197    response = table.delete_item(198        Key={199            'gatewayId':gatewayId,200            }201        )202    return response203    204def getClient():205    dynamodb = boto3.client('dynamodb' ,region_name='us-east-2')206    dynamodb_table = boto3.resource('dynamodb')207    return dynamodb_table208    209def lambda_handler(event, context):210    # TODO implement211    print("input from mobile-"+ event['body'])212    print("input from type mobile-" )213    print(type(event['body']))214    body = json.loads(event['body'])215    print(type(body))216    217    ret = deleteDevice(body)218    print("received ret from deleteDevice:")219    print(ret)220    print(ret['errorCode'])221    if (ret['errorCode'] != 200) or (ret['db']['statusCode'] != 200):222        return {223            'statusCode': 500,224            'body': json.dumps(ret)225        }226    else:227        return {228            'statusCode': 200,229            'body': json.dumps(ret)...

GatewayHandler.py

Source:GatewayHandler.py

...143        )144        if response['ResponseMetadata']['HTTPStatusCode'] != 201:145            raise RuntimeError146        return response147    def delete_gateway_response(self, response_type):148        try:149            response = self.client.delete_gateway_response(150                restApiId=self.aws['api_id'],151                responseType=response_type152            )153            if not 200 < response['ResponseMetadata']['HTTPStatusCode'] < 299:154                raise RuntimeError155            return response156        except self.client.exceptions.NotFoundException:157            return None158    def has_method(self, resource_id, method):159        try:160            response = self.client.get_method(161                restApiId=self.aws['api_id'],162                resourceId=resource_id,163                httpMethod=method...

Wrapper.py

Source:Wrapper.py

...67                'gatewayresponse.header.Access-Control-Allow-Headers': '\'Content-Type,X-Amz-Date,Authorization,X-Api-Key,X-Amz-Security-Token\'',68                'gatewayresponse.header.Access-Control-Allow-Methods': '\'HEAD,GET,POST,PUT,PATCH,DELETE,ANY,OPTIONS\'',69                'gatewayresponse.header.Access-Control-Allow-Origin': '\'*\''70            }71            self.gateway_handler.delete_gateway_response('DEFAULT_4XX')72            self.gateway_handler.delete_gateway_response('DEFAULT_5XX')73            self.gateway_handler.create_gateway_response('DEFAULT_4XX', gateway_response_params)74            self.gateway_handler.create_gateway_response('DEFAULT_5XX', gateway_response_params)75    def __create_response(self, resource_id, method, response_code, status_code, response_parameters):76        error_regex = '^\\{{ "gatewayResponse": true, "status": "error", "type": "{0}", "userMessage": " " }}$'77        regex = error_regex.format(response_code).replace(' ', '(.|\\s)*')78        self.gateway_handler.create_method_response(resource_id, method, status_code, response_parameters)79        self.gateway_handler.create_integration_response(resource_id, method, status_code, response_parameters, regex, get_error_response_template())80    def create_resource(self, path):81        if self.get_resource_id(path):82            return83        # get tree of resources84        parts = path.split('/')[1:]85        parent_resource_id = self.get_resource_id('/')86        for i in range(len(parts) + 1):...

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.