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

How to use sns_subscription_arn method in localstack

Best Python code snippet using localstack_python

PagerDuty-AWS-CloudWatch-SNS-Integration.py

Source:PagerDuty-AWS-CloudWatch-SNS-Integration.py

1import boto32import requests3import json4# Update to match a locally stored AWS profile5session = boto3.session.Session(profile_name='augmenthor')6PDAccountAccessKey ='SwEYWx-i3x24JMZhThxt'7# Update to match your PagerDuty email address8PD_ACCT_OWNER_EMAIL = 'cristian.i.balan@icloud.com'9sns_clnt = session.client('sns')10sns_topic_arn =None11sns_subscription_pd_protocol = 'HTTPS'12sns_subscription_arn =None13def create_escalation_policy(team_id):14    url = 'https://api.pagerduty.com/escalation_policies'15    headers = {16        'Accept': 'application/vnd.pagerduty+json;version=2',17        'Authorization': 'Token token={token}'.format(token=PDAccountAccessKey),18        'Content-type': 'application/json',19        'From': PD_ACCT_OWNER_EMAIL20    }21    payload = {22    "escalation_policy": {23        "type": "escalation_policy",24        "name": "DevOpsAPI",25        "escalation_rules": [26            {27                "escalation_delay_in_minutes": 30,28                "targets": [29                    {30                        "id": "PCHN6LF",31                            "type": "user_reference",32                            "summary": "Ionel Balan",33                            "self": "https://api.pagerduty.com/users/PCHN6LF",34                            "html_url": "https://cbalan.pagerduty.com/users/PCHN6LF"35                    }36                ]37            }38        ],39        "num_loops": 0,40        "on_call_handoff_notifications": "if_has_services",41        "teams": [42            {43                "id": "{teamid}".format(teamid=team_id),44                "type": "team_reference",45                "summary": "ACME Hotel Back Office - DevOps",46                "self": "https://api.pagerduty.com/teams/PJKUNJK",47                "html_url": "https://cbalan.pagerduty.com/teams/PJKUNJK"48            }49        ],50        "description": "Here is the ep for the devops team."51    }52}53    try:54        r = requests.post(url, headers=headers, data=json.dumps(payload))55        print('Status Code: {code}'.format(code=r.status_code))56        print('escalation ploicy id: '+ r.json()['escalation_policy']['id'])57        return r.json()['escalation_policy']['id']58    except:59        print ("Error processing the create_escalation_policy's POST request")60def create_service(escalation_policy_id):61    url = 'https://api.pagerduty.com/services'62    headers = {63        'Accept': 'application/vnd.pagerduty+json;version=2',64        'Authorization': 'Token token={token}'.format(token=PDAccountAccessKey),65        'Content-type': 'application/json',66        'From': PD_ACCT_OWNER_EMAIL67    }68    payload = {69    "service": {70        "type": "service",71        "name": "AWS CloudWatch API",72        "description": "AWS CloudWatch service (created via API)",73        "status": "active",74        "escalation_policy": {75            "id": escalation_policy_id,76            "type": "escalation_policy_reference"77        },78        "incident_urgency_rule": {79            "type": "constant",80            "urgency": "high"81        },82        "alert_creation": "create_alerts_and_incidents"83    }84}85    try:86        r = requests.post(url, headers=headers, data=json.dumps(payload))87        print('Status Code: {code}'.format(code=r.status_code))88        print('service id: '+r.json()['service']['id'])89        return r.json()['service']['id']90    except:91        print ("Error processing the create_service's POST request")92def create_integration(service_id, subdomain):93    url = 'https://api.pagerduty.com/services/{id}/integrations'.format(id=service_id)94    headers = {95        'Accept': 'application/vnd.pagerduty+json;version=2',96        'Authorization': 'Token token={token}'.format(token=PDAccountAccessKey),97        'Content-type': 'application/json',98        'From': PD_ACCT_OWNER_EMAIL99    }100    payload = {101        "integration": {102            "type": "event_transformer_api_inbound_integration",103            "name": "Amazon CloudWatch API",104            "service": {105                "id": service_id,106                "type": "service_reference",107                "summary": "Amazon CloudWatch API",108                "self": "https://api.pagerduty.com/services/{id}".format(id=service_id),109                "html_url": "https://{sd}.pagerduty.com/services/{id}".format(sd=subdomain,id=service_id)110            },111            "vendor": {112              "id": "PZQ6AUS",113              "type": "vendor_reference"114            }115        }116    }117    try:118        r = requests.post(url, headers=headers, data=json.dumps(payload))119        print('Status Code: {code}'.format(code=r.status_code))120        print('integration key: '+r.json()['integration']['integration_key'])121        return r.json()['integration']['integration_key']122    except Exception as error:123        print(error)124def create_sns_topic(topicname):125    # create an SNS topic126    try:127        response = sns_clnt.create_topic(Name=topicname)128        return response['TopicArn']129    except Exception as error:130        print(error)131def create_sns_subcription_endpoint(snstopicarn):132    # Subcribe to a PD endpoint (previously created in PD's admin console)133    try:134        response = sns_clnt.subscribe(135            TopicArn=snstopicarn,136            Protocol=sns_subscription_pd_protocol,137            Endpoint=sns_subscription_pd_endpoint,138            ReturnSubscriptionArn=True139        )140        return response['SubscriptionArn']141    except Exception as error:142        print(error)143if __name__ == '__main__':144    #create in PagerDuty sbdomain:145    # - an escalation policy for a single team, a new service146    # - a new service which will create alerts and incidents147    # - a new integration with AWS CloudWatch for the service created148    ep_team_id="PX7BN8L"149    service_subdomain="cbalan"150    ik=create_integration(create_service(create_escalation_policy(ep_team_id)), service_subdomain)151    #use the integration key to create an AWS SNS topic and subscription152    sns_subscription_pd_endpoint = "https://events.pagerduty.com/integration/{integration_key}/enqueue".format(153        integration_key=ik)154    #CloudWatch to pagerDuty Topic and Subscription creation155    sns_topic_arn=create_sns_topic("CloudWatch-to-PagerDuty")156    print("sns_topic_arn: {}".format(sns_topic_arn))157    sns_subscription_arn= create_sns_subcription_endpoint(sns_topic_arn)...

sns.py

Source:sns.py

...73        # TODO: use get_subscription_attributes to compare FilterPolicy74        return result[0] if result else None75    @staticmethod76    def get_deploy_templates():77        def sns_subscription_arn(params, resources, resource_id, **kwargs):78            resource = resources[resource_id]79            return resource["PhysicalResourceId"]80        def sns_subscription_params(params, **kwargs):81            def attr_val(val):82                return json.dumps(val) if isinstance(val, (dict, list)) else str(val)83            attrs = [84                "DeliveryPolicy",85                "FilterPolicy",86                "RawMessageDelivery",87                "RedrivePolicy",88            ]89            result = dict([(a, attr_val(params[a])) for a in attrs if a in params])90            return result91        return {...

Blogs

Check out the latest blogs from LambdaTest on this topic:

How To Automate Toggle Buttons In Selenium Java

If you pay close attention, you’ll notice that toggle switches are all around us because lots of things have two simple states: either ON or OFF (in binary 1 or 0).

Why Agile Teams Have to Understand How to Analyze and Make adjustments

How do we acquire knowledge? This is one of the seemingly basic but critical questions you and your team members must ask and consider. We are experts; therefore, we understand why we study and what we should learn. However, many of us do not give enough thought to how we learn.

QA Innovation – Using the senseshaping concept to discover customer needs

QA Innovation - Using the senseshaping concept to discover customer needs QA testers have a unique role and responsibility to serve the customer. Serving the customer in software testing means protecting customers from application defects, failures, and perceived failures from missing or misunderstood requirements. Testing for known requirements based on documentation or discussion is the core of the testing profession. One unique way QA testers can both differentiate themselves and be innovative occurs when senseshaping is used to improve the application user experience.

Best 23 Web Design Trends To Follow In 2023

Having a good web design can empower business and make your brand stand out. According to a survey by Top Design Firms, 50% of users believe that website design is crucial to an organization’s overall brand. Therefore, businesses should prioritize website design to meet customer expectations and build their brand identity. Your website is the face of your business, so it’s important that it’s updated regularly as per the current web design trends.

Developers and Bugs – why are they happening again and again?

Entering the world of testers, one question started to formulate in my mind: “what is the reason that bugs happen?”.

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.