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

How to use send_event_to_api_destination method in localstack

Best Python code snippet using localstack_python

message_forwarding.py

Source:message_forwarding.py

...76            Record={"Data": to_bytes(json.dumps(event))},77        )78    elif ":events:" in target_arn:79        if ":api-destination/" in target_arn or ":destination/" in target_arn:80            send_event_to_api_destination(target_arn, event)81        else:82            events_client = connect_to_service("events", region_name=region)83            eventbus_name = target_arn.split(":")[-1].split("/")[-1]84            events_client.put_events(85                Entries=[86                    {87                        "EventBusName": eventbus_name,88                        "Source": event.get("source"),89                        "DetailType": event.get("detail-type"),90                        "Detail": event.get("detail"),91                    }92                ]93            )94    elif ":kinesis:" in target_arn:95        partition_key_path = dict_utils.get_safe(96            target_attributes,97            "$.KinesisParameters.PartitionKeyPath",98            default_value="$.id",99        )100        stream_name = target_arn.split("/")[-1]101        partition_key = dict_utils.get_safe(event, partition_key_path, event["id"])102        kinesis_client = connect_to_service("kinesis", region_name=region)103        kinesis_client.put_record(104            StreamName=stream_name,105            Data=to_bytes(json.dumps(event)),106            PartitionKey=partition_key,107        )108    elif ":logs:" in target_arn:109        log_group_name = target_arn.split(":")[-1]110        logs_client = connect_to_service("logs", region_name=region)111        log_stream_name = str(uuid.uuid4())112        logs_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)113        logs_client.put_log_events(114            logGroupName=log_group_name,115            logStreamName=log_stream_name,116            logEvents=[{"timestamp": now_utc(millis=True), "message": json.dumps(event)}],117        )118    else:119        LOG.warning('Unsupported Events rule target ARN: "%s"' % target_arn)120def send_event_to_api_destination(target_arn, event):121    """Send an event to an EventBridge API destination122    See https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-api-destinations.html"""123    # ARN format: ...:api-destination/{name}/{uuid}124    region = target_arn.split(":")[3]125    api_destination_name = target_arn.split(":")[-1].split("/")[1]126    events_client = connect_to_service("events", region_name=region)127    destination = events_client.describe_api_destination(Name=api_destination_name)128    # get destination endpoint details129    method = destination.get("HttpMethod", "GET")130    endpoint = destination.get("InvocationEndpoint")131    state = destination.get("ApiDestinationState") or "ACTIVE"132    LOG.debug('Calling EventBridge API destination (state "%s"): %s %s' % (state, method, endpoint))133    headers = {134        # default headers AWS sends with every api destination call...

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.