Best Python code snippet using localstack_python
monitor.py
Source:monitor.py
1# -*- coding: UTF-8 â*-2"""3@Author: LennonChin4@Contact: i@coderap.com5@Date: 2021-10-196"""7import sys8import os9import random10import datetime11import requests12import json13import time14import hmac15import hashlib16import base6417import urllib.parse18class Utils:19 @staticmethod20 def time_title(message):21 return "[{}] {}".format(datetime.datetime.now().strftime('%H:%M:%S'), message)22 @staticmethod23 def log(message):24 print(Utils.time_title(message))25 @staticmethod26 def send_message(notification_configs, message, **kwargs):27 if len(message) == 0:28 return29 # Wrapper for exception caught30 def invoke(func, configs):31 try:32 func(configs, message, **kwargs)33 except Exception as err:34 Utils.log(err)35 # DingTalk message36 invoke(Utils.send_dingtalk_message, notification_configs["dingtalk"])37 # Bark message38 invoke(Utils.send_bark_message, notification_configs["bark"])39 # Telegram message40 invoke(Utils.send_telegram_message, notification_configs["telegram"])41 @staticmethod42 def send_dingtalk_message(dingtalk_configs, message, **kwargs):43 if len(dingtalk_configs["access_token"]) == 0 or len(dingtalk_configs["secret_key"]) == 0:44 return45 timestamp = str(round(time.time() * 1000))46 secret_enc = dingtalk_configs["secret_key"].encode('utf-8')47 string_to_sign = '{}\n{}'.format(timestamp, dingtalk_configs["secret_key"])48 string_to_sign_enc = string_to_sign.encode('utf-8')49 hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()50 sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))51 headers = {52 'Content-Type': 'application/json'53 }54 params = {55 "access_token": dingtalk_configs["access_token"],56 "timestamp": timestamp,57 "sign": sign58 }59 content = {60 "msgtype": "text" if "message_type" not in kwargs else kwargs["message_type"],61 "text": {62 "content": message63 }64 }65 response = requests.post("https://oapi.dingtalk.com/robot/send", headers=headers, params=params, json=content)66 Utils.log("Dingtalkéåºæ¶æ¯çæ
ï¼{}".format(response.status_code))67 @staticmethod68 def send_telegram_message(telegram_configs, message, **kwargs):69 if len(telegram_configs["bot_token"]) == 0 or len(telegram_configs["chat_id"]) == 0:70 return71 headers = {72 'Content-Type': 'application/json'73 }74 proxies = {75 "https": telegram_configs["http_proxy"],76 }77 content = {78 "chat_id": telegram_configs["chat_id"],79 "text": message80 }81 url = "https://api.telegram.org/bot{}/sendMessage".format(telegram_configs["bot_token"])82 response = requests.post(url, headers=headers, proxies=proxies, json=content)83 Utils.log("Telegraméåºæ¶æ¯çæ
ï¼{}".format(response.status_code))84 @staticmethod85 def send_bark_message(bark_configs, message, **kwargs):86 if len(bark_configs["url"]) == 0:87 return88 url = "{}/{}".format(bark_configs["url"], urllib.parse.quote(message, safe=""))89 response = requests.post(url, params=bark_configs["query_parameters"])90 Utils.log("Barkéåºæ¶æ¯çæ
ï¼{}".format(response.status_code))91class AppleStoreMonitor:92 headers = {93 'sec-ch-ua': '"Google Chrome";v="93", " Not;A Brand";v="99", "Chromium";v="93"',94 'Referer': 'https://www.apple.com.cn/store',95 'DNT': '1',96 'sec-ch-ua-mobile': '?0',97 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.82 Safari/537.36',98 'sec-ch-ua-platform': '"macOS"',99 }100 def __init__(self):101 self.count = 1102 @staticmethod103 def config():104 """105 é²è¡åé¡æä½106 """107 products = json.load(open('products.json', encoding='utf-8'))108 configs = {109 "selected_products": {},110 "selected_area": "",111 "exclude_stores": [],112 "notification_configs": {113 "dingtalk": {114 "access_token": "",115 "secret_key": ""116 },117 "telegram": {118 "bot_token": "",119 "chat_id": "",120 "http_proxy": ""121 },122 "bark": {123 "url": "",124 "query_parameters": {125 "url": None,126 "isArchive": None,127 "group": None,128 "icon": None,129 "automaticallyCopy": None,130 "copy": None131 }132 }133 },134 "scan_interval": 30,135 "alert_exception": False136 }137 while True:138 # chose product type139 print('--------------------')140 for index, item in enumerate(products):141 print('[{}] {}'.format(index, item))142 product_type = list(products)[int(input('é¸æè¦æ¥çåèï¼'))]143 # chose product classification144 print('--------------------')145 for index, (key, value) in enumerate(products[product_type].items()):146 print('[{}] {}'.format(index, key))147 product_classification = list(products[product_type])[int(input('é¸æè¦æ¥çåèåé¡ï¼'))]148 # chose product classification149 print('--------------------')150 for index, (key, value) in enumerate(products[product_type][product_classification].items()):151 print('[{}] {}'.format(index, value))152 product_model = list(products[product_type][product_classification])[int(input('é¸æè¦æ¥çIPHONEåèï¼'))]153 configs["selected_products"][product_model] = (154 product_classification, products[product_type][product_classification][product_model])155 print('--------------------')156 if len(input('æ¯å¦å¢å æ´ç¢å[Enterç¹¼çºæ·»å ï¼éEnteréµéåº]ï¼')) != 0:157 break158 # config area159 print('é¸æé ç´å°åï¼')160 url_param = ['state', 'city', 'district']161 choice_params = {}162 param_dict = {}163 for step, param in enumerate(url_param):164 print('è«ç¨å¾...{}/{}'.format(step + 1, len(url_param)))165 response = requests.get("https://www.apple.com.cn/shop/address-lookup", headers=AppleStoreMonitor.headers,166 params=choice_params)167 result_param = json.loads(response.text)['body'][param]168 if type(result_param) is dict:169 result_data = result_param['data']170 print('--------------------')171 for index, item in enumerate(result_data):172 print('[{}] {}'.format(index, item['value']))173 input_index = int(input('è«é¸æåèï¼'))174 choice_result = result_data[input_index]['value']175 param_dict[param] = choice_result176 choice_params[param] = param_dict[param]177 else:178 choice_params[param] = result_param179 print('æ£å¨è¼å
¥ç¶²çµ¡è³æº...')180 response = requests.get("https://www.apple.com.cn/shop/address-lookup", headers=AppleStoreMonitor.headers,181 params=choice_params)182 selected_area = json.loads(response.text)['body']['provinceCityDistrict']183 configs["selected_area"] = selected_area184 print('--------------------')185 print("é¸æé ç´å°åæ¯ï¼{}ï¼è¼å
¥é ç´å°æ¢å¨åç´çåº...".format(selected_area))186 store_params = {187 "location": selected_area,188 "parts.0": list(configs["selected_products"].keys())[0]189 }190 response = requests.get("https://www.apple.com.cn/shop/fulfillment-messages",191 headers=AppleStoreMonitor.headers, params=store_params)192 stores = json.loads(response.text)['body']["content"]["pickupMessage"]["stores"]193 for index, store in enumerate(stores):194 print("[{}] {}ï¼å°åï¼{}".format(index, store["storeName"], store["retailStore"]["address"]["street"]))195 exclude_stores_indexes = input('æé¤ç¡éç£æ¸¬çç´çåºï¼è¼¸å
¥åºèè¾[ç´æ¥Enter代表å
¨é¨ç£æ¸¬ï¼å¤ååºåºèä»¥ç©ºæ ¼åé]ï¼').strip().split()196 if len(exclude_stores_indexes) != 0:197 print("å·²é¸æçç¡éç£æ¸¬ç´åºï¼{}".format("ï¼".join(list(map(lambda i: stores[int(i)]["storeName"], exclude_stores_indexes)))))198 configs["exclude_stores"] = list(map(lambda i: stores[int(i)]["storeNumber"], exclude_stores_indexes))199 print('--------------------')200 # config notification configurations201 notification_configs = configs["notification_configs"]202 # config dingtalk notification203 dingtalk_access_token = input('è¾å
¥ééæ©å¨äººAccess Token[å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼')204 dingtalk_secret_key = input('è¾å
¥ééæ©å¨äººSecret Key[å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼')205 # write dingtalk configs206 notification_configs["dingtalk"]["access_token"] = dingtalk_access_token207 notification_configs["dingtalk"]["secret_key"] = dingtalk_secret_key208 # config telegram notification209 print('--------------------')210 telegram_chat_id = input('è¾å
¥Telegramæºå¨äººChat ID[å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼')211 telegram_bot_token = input('è¾å
¥Telegramæºå¨äººToken[å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼')212 telegram_http_proxy = input('è¾å
¥Telegram HTTP代çå°å[å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼')213 # write telegram configs214 notification_configs["telegram"]["chat_id"] = telegram_chat_id215 notification_configs["telegram"]["bot_token"] = telegram_bot_token216 notification_configs["telegram"]["http_proxy"] = telegram_http_proxy217 # config bark notification218 print('--------------------')219 bark_url = input('è¾å
¥Bark URL[å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼')220 # write dingtalk configs221 notification_configs["bark"]["url"] = bark_url222 # 輸å
¥æçééæé223 print('--------------------')224 configs["scan_interval"] = int(input('è¾å
¥æçééæé[以ç§çºå®ä½ï¼é»èªçº30ç§ï¼å¦ä¸é
ç½®ç´æ¥Enterå³å¯]ï¼') or 30)225 # æ¯å¦å°ç°å¸¸é²è¡è¦å226 print('--------------------')227 configs["alert_exception"] = (input('æ¯å¦å¨ç¨åºç°å¸¸æç¼åºéç¥[Y/nï¼é»èªçºn]ï¼').lower().strip() or "n") == "y"228 with open('apple_store_monitor_configs.json', 'w') as file:229 json.dump(configs, file, indent=2)230 print('--------------------')231 print("æçé
置己çæï¼ä¸¦å·²å¯«å
¥å°{}æ件ä¸\nè¯·ä½¿ç¨ python {} start å½ä»¤å§åç£æ¸¬".format(file.name, os.path.abspath(__file__)))232 def start(self):233 """234 éå§ç£æ¸¬235 """236 configs = json.load(open('apple_store_monitor_configs.json', encoding='utf-8'))237 selected_products = configs["selected_products"]238 selected_area = configs["selected_area"]239 exclude_stores = configs["exclude_stores"]240 notification_configs = configs["notification_configs"]241 scan_interval = configs["scan_interval"]242 alert_exception = configs["alert_exception"]243 products_info = []244 for index, product_info in enumerate(selected_products.items()):245 products_info.append("ã{}ã{}".format(index, " ".join(product_info[1])))246 message = "æºåéå§ç£æ¸¬ï¼ååä¿¡æ¯å¦ä¸ï¼\n{}\nå貨ååï¼{}\næçé »æ¬¡ï¼{}ç§/次".format("\n".join(products_info), selected_area,247 scan_interval)248 Utils.log(message)249 Utils.send_message(notification_configs, message)250 params = {251 "location": selected_area,252 "mt": "regular",253 }254 code_index = 0255 product_codes = selected_products.keys()256 for product_code in product_codes:257 params["parts.{}".format(code_index)] = product_code258 code_index += 1259 # ä¸æ¬¡æ´é»éç¥æé260 last_exactly_time = -1261 while True:262 available_list = []263 tm_hour = time.localtime(time.time()).tm_hour264 try:265 # æ´æ°è«æ±æé266 params["_"] = int(time.time() * 1000)267 response = requests.get("https://www.apple.com.cn/shop/fulfillment-messages",268 headers=AppleStoreMonitor.headers,269 params=params)270 json_result = json.loads(response.text)271 stores = json_result['body']['content']['pickupMessage']['stores']272 Utils.log(273 '-------------------- 第{}次æç --------------------'.format(274 self.count + 1))275 for item in stores:276 store_name = item['storeName']277 if item["storeNumber"] in exclude_stores:278 print("ã{}ï¼å·²æé¤ã".format(store_name))279 continue280 print("{:-<100}".format("ã{}ã".format(store_name)))281 for product_code in product_codes:282 pickup_search_quote = item['partsAvailability'][product_code]['pickupSearchQuote']283 pickup_display = item['partsAvailability'][product_code]['pickupDisplay']284 store_pickup_product_title = item['partsAvailability'][product_code]['storePickupProductTitle']285 print('\tã{}ã{}'.format(pickup_search_quote, store_pickup_product_title))286 if pickup_search_quote == 'ä»å¤©å¯å貨' or pickup_display != 'unavailable':287 available_list.append((store_name, product_code, store_pickup_product_title))288 if len(available_list) > 0:289 messages = []290 print("å½ä¸è²¨æºï¼è«æ³¨æ >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")291 Utils.log("以ä¸ç´çåºé ç´å¯ç¨ï¼")292 for item in available_list:293 messages.append("ã{}ã {}".format(item[0], item[2]))294 print("ã{}ã{}".format(item[0], item[2]))295 print("å½ä¸è²¨æºï¼è«æ³¨æ >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")296 Utils.send_message(notification_configs,297 Utils.time_title(298 "第{}次æçå°ç´çåºæ貨ï¼ä¿¡æ¯å¦ä¸ï¼\n{}".format(self.count, "\n".join(messages))))299 except Exception as err:300 Utils.log(err)301 # 6:00 ~ 23:00æåéå¼å¸¸æ¶æ¯302 if alert_exception and 6 <= tm_hour <= 23:303 Utils.send_message(notification_configs,304 Utils.time_title("第{}次æçå°åºç¾ç°å¸¸ï¼{}".format(self.count, repr(err))))305 if len(available_list) == 0:306 interval = max(random.randint(int(scan_interval / 2), scan_interval * 2), 5)307 Utils.log('{}ç§å¾é²è¡ç¬¬{}次å試...'.format(interval, self.count))308 # æ´ç¹éç¥ï¼ç¨äºé¶æ®µæ§æ£æµåºç¨æ¯å¦æ£å¸¸309 if last_exactly_time != tm_hour and (6 <= tm_hour <= 23):310 Utils.send_message(notification_configs,311 Utils.time_title("å·²æç{}次ï¼æçç¨åºéä½æ£å¸¸".format(self.count)))312 last_exactly_time = tm_hour313 time.sleep(interval)314 else:315 time.sleep(5)316 # 次æ°èªå¢317 self.count += 1318if __name__ == '__main__':319 args = sys.argv320 if len(args) != 2:321 print("""322 Usage: python {} <option>323 option can be:324 \tconfig: pre config of products or notification325 \tstart: start to monitor326 """.format(args[0]))327 exit(1)328 if args[1] == "config":329 AppleStoreMonitor.config()330 if args[1] == "start":...
notification_manager.py
Source:notification_manager.py
1from abc import ABC, abstractmethod2from collections import namedtuple3import asyncio4from threading import Thread5from pathlib import Path6import time7from typing import List, Any, Optional, Callable8import yaml9from sensor_manager import SensorManager10from email_manager import EmailManager11class NotificationConfig(ABC):12 def __init__(self, get_time: Callable[[], float], sender_email: str, receiver_email: str, message_subject: str, message: Optional[str], check_interval: float):13 self.get_time = get_time14 self.sender_email = sender_email15 self.receiver_email = receiver_email16 self.message_subject = message_subject17 self.message = message18 self.check_interval = check_interval19 @classmethod20 @abstractmethod21 def get_type_name(cls):22 pass23 @classmethod24 def from_dict(cls, raw_dict, get_time):25 """get_time: function to get current time in seconds from epoch"""26 if raw_dict['type'] != cls.get_type_name():27 raise Exception('raw dictionary does not seem to encode for this message type')28 cls_params = cls.params_from_dict(raw_dict)29 if cls_params == False:30 raise Exception('raw dictionary does not seem to encode for this message type')31 32 result = cls(33 get_time=get_time,34 sender_email=raw_dict['sender'],35 receiver_email=raw_dict['receiver'],36 message_subject=raw_dict['message_subject'],37 message=raw_dict['message'] if 'message' in raw_dict else None,38 check_interval=float(raw_dict['check_interval']),39 **cls_params)40 return result41 @classmethod42 @abstractmethod43 def params_from_dict(cls, raw_dict):44 pass45 @abstractmethod46 def check_signal(self, current_sensor_values) -> bool:47 pass48class BaseThresholdNotificationConfig(NotificationConfig):49 def __init__(self, sensor_id: str, threshold: float, min_breach_duration: float, **kwargs):50 super().__init__(**kwargs)51 self.sensor_id = sensor_id52 self.threshold = threshold53 self.min_breach_duration = min_breach_duration54 self.previous_check_value = None55 self.tracking_breach = False56 self.current_breach_start_timestamp = None57 self.current_breach_notified = False58 @classmethod59 @abstractmethod60 def get_type_name(cls):61 pass62 @classmethod63 def params_from_dict(cls, raw_dict):64 return {65 'sensor_id': raw_dict['sensor'],66 'threshold': float(raw_dict['threshold']),67 'min_breach_duration': float(raw_dict['min_breach_duration'])68 }69 def check_signal(self, current_sensor_values) -> bool:70 breached = False71 72 try:73 breached = self.check_threshold_breached(current_sensor_values)74 except Exception as e:75 breached = self.tracking_breach76 if breached and not self.tracking_breach:77 self.tracking_breach = True78 self.current_breach_start_timestamp = self.get_time()79 self.current_breach_notified = False80 elif not breached and self.tracking_breach:81 self.tracking_breach = False82 if breached and self.tracking_breach:83 current_timestamp = self.get_time()84 duration = current_timestamp - self.current_breach_start_timestamp85 duration_reached = duration >= self.min_breach_duration86 result = duration_reached and not self.current_breach_notified 87 if duration_reached:88 self.current_breach_notified = True89 90 return result91 return False92 @abstractmethod93 def check_threshold_breached(self, current_sensor_values):94 pass95 def __str__(self):96 return f'Notification {{ {self.get_type_name()}: {str(self.threshold)}, sender: {self.sender_email}, receiver: {self.receiver_email} }}'97class FallBelowNotificationConfig(BaseThresholdNotificationConfig):98 @classmethod99 def get_type_name(cls):100 return 'fall_below'101 102 def check_threshold_breached(self, current_sensor_values) -> bool:103 current_value = current_sensor_values[self.sensor_id]104 result = False105 if self.previous_check_value is None:106 result = self.threshold > current_value107 else:108 result = self.threshold > current_value and self.previous_check_value > self.threshold109 self.previous_check_value = current_value110 return result111class RiseAboveNotificationConfig(BaseThresholdNotificationConfig):112 @classmethod113 def get_type_name(cls):114 return 'rise_above'115 def check_threshold_breached(self, current_sensor_values) -> bool:116 current_value = current_sensor_values[self.sensor_id]117 if current_value is None:118 print('delaying check because sensor read value is none')119 raise Exception('no valid value, could not check threshold breached')120 else:121 return current_value > self.threshold122class SystemStartNotificationConfig(NotificationConfig):123 def __init__(self, *args, **kwargs):124 super().__init__(*args, **kwargs)125 self.checked_once = False126 @classmethod127 def params_from_dict(cls, raw_dict):128 return {}129 @classmethod130 def get_type_name(cls):131 return 'system_start'132 def check_signal(self, current_sensor_values) -> bool:133 result = not self.checked_once134 self.checked_once = True135 return result136class NotificationManager:137 def __init__(self, notification_configs, sensor_manager: SensorManager, email_manager: EmailManager, get_time = time.time):138 self.notification_configs = notification_configs139 self.sensor_manager = sensor_manager140 self.email_manager = email_manager141 self.get_time = get_time142 self.last_check_timestamps = {}143 self.current_sensor_values = None144 self.master_check_interval = 1 if len(notification_configs) == 0 else min([x.check_interval for x in notification_configs])145 self.watch_thread = None146 print('Notification Manager master_check_interval', self.master_check_interval)147 def start_watching(self):148 if self.watch_thread is not None:149 raise Exception('can only start one watching thread per notification manager')150 self.watch_thread = Thread(target=self.watch_loop)151 self.watch_thread.start()152 def watch_loop(self):153 while True:154 self.single_watch_step()155 time.sleep(self.master_check_interval)156 def single_watch_step(self):157 current_timestamp = self.get_time()158 self.current_sensor_values = asyncio.run(self.sensor_manager.get_latest_values_filled_with_previous())159 try:160 for index, notification_config in enumerate(self.notification_configs):161 if not index in self.last_check_timestamps or\162 self.last_check_timestamps[index] + notification_config.check_interval <= current_timestamp:163 164 self.last_check_timestamps[index] = current_timestamp165 try:166 if notification_config.check_signal(self.current_sensor_values):167 self.send_notification(notification_config)168 except Exception as e:169 print(f'an error occurred while attempting to check or send notification {notification_config}', e)170 except Exception as e:171 print('an error occurred while checking for possible notifications', e)172 def send_notification(self, notification_config: NotificationConfig):173 print('send notification for config:', notification_config)174 175 text = notification_config.message176 if text is None:177 text = 'notification was sent automatically'178 self.email_manager.send_email(179 sender_address=notification_config.sender_email,180 receiver_address=notification_config.receiver_email,181 subject=notification_config.message_subject,182 text=text)183 @staticmethod184 def parse_notifications_config(config_file_path: Path, get_time=time.time) -> List[Any]:185 with open(config_file_path, 'rb') as file:186 raw_notification_configs = yaml.load(file, Loader=yaml.Loader)187 potential_config_types = [FallBelowNotificationConfig, RiseAboveNotificationConfig, SystemStartNotificationConfig]188 notification_configs = []189 for index, raw_notification_config in enumerate(raw_notification_configs):190 found_matching_type = False191 for potential_type in potential_config_types:192 try:193 parsed = potential_type.from_dict(raw_notification_config, get_time=get_time)194 if parsed is not None:195 notification_configs.append(parsed)196 found_matching_type = True197 break 198 except Exception as e:199 print(f'error when parsing config {index}: {e}')200 pass201 if not found_matching_type:202 raise Exception(f'found no matching type for notification config at index: {index}, raw: {raw_notification_config}')203 ...
asg_event_notifications_util.py
Source:asg_event_notifications_util.py
1from __future__ import absolute_import2from __future__ import print_function3import boto34import click5@click.group()6def cli():7 pass8def get_asg_infos():9 response = client.describe_auto_scaling_groups(MaxRecords=100)10 auto_scaling_groups = response['AutoScalingGroups']11 return auto_scaling_groups12def get_asg_names():13 asg_names = list()14 for asg in get_asg_infos():15 asg_names.append(asg['AutoScalingGroupName'])16 return asg_names17def get_asg_event_notifications(asg):18 event_notifications = list()19 response = \20 client.describe_notification_configurations(AutoScalingGroupNames=[asg],21 MaxRecords=100)22 notification_configs = response['NotificationConfigurations']23 for notification in notification_configs:24 event_notifications.append(notification['NotificationType'])25 return event_notifications26@click.command()27def show_asg_event_notifications():28 try:29 for asg in get_asg_names():30 event_notifications = get_asg_event_notifications(asg)31 if event_notifications:32 print(("Event notifications: {0} are set for ASG: {1}".format(event_notifications,33 asg)))34 else:35 print(("No Event Notifications found for ASG {}".format(asg)))36 except Exception as e:37 print(e)38@click.command()39@click.option('--topic_arn', help='The ARN of Amazon SNS topic',40 required=True)41@click.option('--event',42 help='The type of event that causes the notification to be sent'43 , default='autoscaling:EC2_INSTANCE_LAUNCH_ERROR')44@click.option('--confirm', default=False, required=False, is_flag=True,45 help='Set this to create event notification for asg')46def create_asg_event_notifications(47 topic_arn,48 event,49 confirm,50 ):51 asg_names = get_asg_names()52 asg_to_create_event_notifications = list()53 for asg_name in asg_names:54 event_notifications = get_asg_event_notifications(asg_name)55 if event in event_notifications:56 continue57 else:58 asg_to_create_event_notifications.append(asg_name)59 if confirm is False:60 print(("Would have created the event notification for asgs {}".format(asg_to_create_event_notifications)))61 else:62 try:63 for asg in asg_to_create_event_notifications:64 response = \65 client.put_notification_configuration(AutoScalingGroupName=asg,66 TopicARN=topic_arn, NotificationTypes=[event])67 print(("Created {0} event notifications for auto scaling group {1}").format(event,68 asg))69 except Exception as e:70 print(e)71cli.add_command(show_asg_event_notifications)72cli.add_command(create_asg_event_notifications)73if __name__ == '__main__':74 client = boto3.client('autoscaling')...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!