KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use list_endpoints_by_platform_application method in localstack

Best Python code snippet using localstack_python

sns.py

Source:sns.py

...65    return client.get_topic_attributes(**kwargs)['Attributes']66@sts_conn('sns')67@paginated('Endpoints', request_pagination_marker="NextToken", response_pagination_marker="NextToken")68@rate_limited()69def list_endpoints_by_platform_application(client=None, **kwargs):70    return client.list_endpoints_by_platform_application(**kwargs)71@sts_conn('sns')72@paginated('phoneNumbers', request_pagination_marker="nextToken", response_pagination_marker="nextToken")73@rate_limited()74def list_phone_numbers_opted_out(client=None, **kwargs):75    return client.list_phone_numbers_opted_out(**kwargs)76@sts_conn('sns')77@paginated('PlatformApplications', request_pagination_marker="NextToken", response_pagination_marker="NextToken")78@rate_limited()79def list_platform_applications(client=None, **kwargs):80    return client.list_platform_applications(**kwargs)81@sts_conn('sns')82@paginated('Subscriptions', request_pagination_marker="NextToken", response_pagination_marker="NextToken")83@rate_limited()84def list_subscriptions(client=None, **kwargs):...

tasks.py

Source:tasks.py

...16        app_arn = settings.AWS_SNS_GCM_ARN17    else:18        return19    conn = get_connection_sns()20    res = conn.list_endpoints_by_platform_application(platform_application_arn=app_arn)21    while res.get('ListEndpointsByPlatformApplicationResponse').get('ListEndpointsByPlatformApplicationResult').get(22            'NextToken'):23        endpoints = res.get('ListEndpointsByPlatformApplicationResponse').get(24            'ListEndpointsByPlatformApplicationResult').get('Endpoints')25        for ep in endpoints:26            endpoint_arn = ep.get('EndpointArn')27            token = ep.get('Attributes').get('Token')28            if ep.get('Attributes').get('Enabled') == 'true':29                objs = SNSToken.objects.filter(registration_id=token).exclude(arn=endpoint_arn)30                for o in objs:31                    conn.delete_endpoint(o.arn)32                    o.delete()33            else:34                conn.delete_endpoint(endpoint_arn)35                SNSToken.objects.filter(arn=endpoint_arn).delete()36        res = conn.list_endpoints_by_platform_application(platform_application_arn=app_arn, next_token=res.get(37            'ListEndpointsByPlatformApplicationResponse').get('ListEndpointsByPlatformApplicationResult').get(38            'NextToken'))39def prune_user_tokens(user):40    tokens = SNSToken.objects.filter(user=user)41    if len(tokens) > 1:42        conn = get_connection_sns()43        registred_ids = []44        for t in tokens:45            attr = conn.get_endpoint_attributes(t.arn).get('GetEndpointAttributesResponse').get(46                'GetEndpointAttributesResult').get('Attributes')47            if attr.get('Enabled') == 'true':48                if attr.get('Token') in registred_ids:49                    conn.delete_endpoint(t.arn)50                    t.delete()...

EndpointsToTopic.py

Source:EndpointsToTopic.py

...7nextToken = None8batch = 09while True:10    if nextToken:11        response = client.list_endpoints_by_platform_application(PlatformApplicationArn=platformApplicationArn, NextToken=nextToken)12    else:13        response = client.list_endpoints_by_platform_application(PlatformApplicationArn=platformApplicationArn)14    endpoints = response['Endpoints']15    batch += len(endpoints)16    print 'batch: %d nextToken: %s' % (batch, nextToken)17    # iterate elements18    print 'end point back size %d' % len(endpoints)19    for item in endpoints:20        endpointArn = item['EndpointArn']21        print endpointArn22        # subscribe23        subResponse = client.subscribe(24            TopicArn=topicArn,25            Protocol='application',26            Endpoint=endpointArn27        )...

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.