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

How to use delete_event_source_mapping method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

...102        EventSourceArn=oldSqsQueueArn,103        FunctionName=oldLambdaArn104    )105    for event_source in event_sources['EventSourceMappings']:106        lambda_client.delete_event_source_mapping(107            UUID=event_source["UUID"]108        )109    print("Linking SQS Events To Lambda")110    lambda_linking_response = lambda_client.create_event_source_mapping(111        EventSourceArn=sqsQueueArn,112        FunctionName=lambdaArn,113    )114    print("Linking SQS Events To Lambda Complete")115    print(lambda_linking_response)116    return { "Status": "SUCCESS" }117def on_delete(event):118    physical_id = event["PhysicalResourceId"]119    props = event["ResourceProperties"]120    print("delete resource %s" % physical_id)121    122    bucketArn = props["bucketArn"]123    bucketName = props["bucketName"]124    sqsQueueArn = props["sqsQueueArn"]125    lambdaArn = props["lambdaArn"]126    s3_client = boto3.client('s3')127    lambda_client = boto3.client('lambda')128    print("Unlinking Bucket Events To SQS")129    bucket_linking_response = s3_client.put_bucket_notification_configuration(130        Bucket=bucketName,131        NotificationConfiguration={132            "QueueConfigurations": []133        }134    )135    print("Unlinking Bucket Events To SQS Complete")136    print(bucket_linking_response)137    print("Deleting Old Linking SQS Events To Lambda")138    event_sources = lambda_client.list_event_source_mappings(139        EventSourceArn=sqsQueueArn,140        FunctionName=lambdaArn141    )142    for event_source in event_sources['EventSourceMappings']:143        lambda_client.delete_event_source_mapping(144            UUID=event_source["UUID"]145        )...

kafka.py

Source:kafka.py

...14        if event["RequestType"] in ["Create", "Update"]:15            if event["RequestType"] == "Update":16                try:17                    print("EventSourceMapping recreation")18                    client.delete_event_source_mapping(UUID=physicalResourceId)19                    while True:20                        client.get_event_source_mapping(UUID=physicalResourceId)21                        time.sleep(10)22                except client.exceptions.ResourceNotFoundException:23                    pass24            response = client.create_event_source_mapping(25                FunctionName=event["ResourceProperties"]["Function"],26                BatchSize=int(event["ResourceProperties"]["BatchSize"]),27                StartingPosition=event["ResourceProperties"]["StartingPosition"],28                Topics=[29                    event["ResourceProperties"]["Topic"]30                ],31                SelfManagedEventSource={32                    "Endpoints": {33                        "KAFKA_BOOTSTRAP_SERVERS": event["ResourceProperties"]["Brokers"]34                    }35                },36                SourceAccessConfigurations=list([37                    {38                        "Type": "VPC_SUBNET",39                        "URI": "subnet:" + subnetId40                    } for subnetId in event["ResourceProperties"]["SubnetIds"]41                ]) + list([42                    {43                        "Type": "VPC_SECURITY_GROUP",44                        "URI": "security_group:" + securityGroupId45                    } for securityGroupId in event["ResourceProperties"]["SecurityGroupIds"]46                ])47            )48            physicalResourceId = response["UUID"]49            while True:50                response = client.get_event_source_mapping(UUID=physicalResourceId)51                if response["State"] in ["Enabled", "Disabled"]:52                    break53                time.sleep(10)54            print("EventSourceMapping successfully created")55        elif event["RequestType"] == "Delete":56            try:57                client.delete_event_source_mapping(UUID=physicalResourceId)58            except client.exceptions.ResourceNotFoundException:59                pass60            print("EventSourceMapping successfully deleted")61    except Exception as exc:62        print("Failed to process:", exc)63        responseStatus = "FAILED"64    finally:...

destroyprep.py

Source:destroyprep.py

...14    )15    print("Destroy all: {0} for stack {1}".format(all, stackname))16    # run python scripts to teardown17    if all:18        delete_event_source_mapping(stackname, lambda_client)19        delete_platform_services(stackname, cloudformation_client, True)20        delete_cf_dists(stackname, cloudfront_client, True)21    else:22        delete_event_source_mapping(stackname, lambda_client)...

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.