关于python 代码结构、函数和调用函数的问题
Question about python code structure, functions and calling functions
我试图创建一个函数来执行以下操作。
(不知道是否值得一提,但这个函数是由另一个函数调用的)
- 使用 boto3 连接到 aws 资源
- 获取 sqs 队列中可用的消息数
- 统计ec2实例数量
- 根据 sqs 队列和 ec2 实例评估一组条件,然后不执行任何操作或写入 sns 主题。
基本上我想在每次 sqs 队列高而消化这些的 ec2 实例数量低时发布到 sns 主题的消息。
import os
import boto3
import logging
import types
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# Create session and clients
sts_client = boto3.client('sts')
sqs_client = boto3.client('sqs')
ec2_client = boto3.client('ec2')
sns_client = boto3.client('sns')
# Call the assume_role method of the STSConnection object and pass the role ARN and a role session name.
assumed_role_object = sts_client.assume_role(
RoleArn=os.environ['ROLE_ARN'],
RoleSessionName="AssumeRoleFromCloudOperations"
)
# From the response that contains the assumed role, get the temporary credentials
credentials = assumed_role_object['Credentials']
assumed_role_session = boto3.Session(
aws_access_key_id=credentials['AccessKeyId'],
aws_secret_access_key=credentials['SecretAccessKey'],
aws_session_token=credentials['SessionToken']
)
# Check the queue size
def sqs():
queue_size = sqs_client.get_queue_attributes(
QueueUrl=os.environ['SQS_QUEUE_URL'],
AttributeNames=['ApproximateNumberOfMessages']
)
messages = int(queue_size["Attributes"]["ApproximateNumberOfMessages"])
return messages
# Count the number of active ec2 instances
def count_instances(ec2):
total_instances = 0
instances = ec2.instances.filter(Filters=[
{
'Instance State': 'instance-state-name',
'Values': ['running'],
'Name': 'tag:Name',
'Values': ['NameOfInstance']
},
])
for _ in instances:
total_instances += 1
return total_instances
print(f"Total number of active scan servers is: {total_instances}")
# Define the SNS Topic which will be integrated with OpsGenie
def sns():
topicArn = os.environ['SNS_ARN']
# Evaluate the set of conditions
def evaluate_conditions(context, event):
sqs()
if messages > int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances > int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS']):
print('False alert')
logger.info()
elif messages < int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances < int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS']):
print('False alert')
logger.info()
elif messages < int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances > int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS']):
print('False alert')
logger.info()
else:
sns.publish(TopicArn=os.environ['SNS_ARN'],
Message='sameple message',
Subject='sample subject')
print("Published to SNS Topic")
处理程序是 handler.evaluate_conditions
我的问题是我怎样才能在这个 lambda 函数中有一些结构?
当我 运行 函数出现命名错误时:
{
"errorMessage": "name 'messages' is not defined",
"errorType": "NameError",
"stackTrace": [
" File \"/var/task/mdc_alert/handler.py\", line 67, in evaluate_conditions\n if messages > int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances > int(\n"
]
}
所以我似乎无法在 evaluate_conditions() 函数中使用消息变量。
如何使“消息”和“total_instances”变量在 evaluate_conditions() 函数中可用?
我完全基于 google 搜索、Whosebug 和 boto3 文档编写了这个函数,因为我没有任何编程经验。
这个结构好不好,还是需要彻底改造?
我是否需要更改函数的顺序,或者创建一个 class?
眼前的问题是 messages
变量未定义。您的 sqs
函数 return 是一个值,但由于您是在 void 上下文中调用它,因此您实际上并没有对该值执行任何操作。您可以通过更改此行来解决此问题:
sqs()
给这个:
messages = sqs()
我还发现 count_instances 函数存在一些问题。它期望收到一个 ec2
变量,但您从 evaluate_conditions 错误地调用了它。您可以将 ec2_client 变量传递给它,也可以直接从函数内部使用 ec2_client 变量。
我建议重命名您的函数以更准确地反映它们的 return 值:
sqs -> sqs_msg_count
count_instances -> running_ec2_count
进行这些更改将使您能够重构 evaluate_conditions 以缩短 if-then 行,使您的代码整体上更易于阅读和遵循。如果您考虑了所有这些建议,您的代码可能如下所示:
# Check the queue size
def sqs_msg_count():
messages = sqs_client.get_queue_attributes(
QueueUrl=os.environ['SQS_QUEUE_URL'],
AttributeNames=['ApproximateNumberOfMessages']
)
return int(messages["Attributes"]["ApproximateNumberOfMessages"])
# Count the number of active ec2 instances
def running_instance_count():
running_instances = ec2_client.instances.filter(Filters=[
{
'Instance State': 'instance-state-name',
'Values': ['running'],
'Name': 'tag:Name',
'Values': ['NameOfInstance']
},
])
return len(running_instances)
# Evaluate the set of conditions
def evaluate_conditions(context, event):
sqs_count = sqs_msg_count()
sqs_average = int(os.environ['AVG_QUEUE_SIZE'])
ec2_count = running_instance_count()
ec2_average = int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS'])
if sqs_count > sqs_average and ec2_count > ec2_average:
print('False alert')
logger.info()
elif sqs_count < sqs_average and ec2_count < ec2_average:
print('False alert')
logger.info()
elif sqs_count < sqs_average and ec2_count > ec2_average:
print('False alert')
logger.info()
else:
sns_client.publish(
TopicArn=os.environ['SNS_ARN'],
Message='sameple message',
Subject='sample subject'
)
print("Published to SNS Topic")
我试图创建一个函数来执行以下操作。 (不知道是否值得一提,但这个函数是由另一个函数调用的)
- 使用 boto3 连接到 aws 资源
- 获取 sqs 队列中可用的消息数
- 统计ec2实例数量
- 根据 sqs 队列和 ec2 实例评估一组条件,然后不执行任何操作或写入 sns 主题。
基本上我想在每次 sqs 队列高而消化这些的 ec2 实例数量低时发布到 sns 主题的消息。
import os
import boto3
import logging
import types
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# Create session and clients
sts_client = boto3.client('sts')
sqs_client = boto3.client('sqs')
ec2_client = boto3.client('ec2')
sns_client = boto3.client('sns')
# Call the assume_role method of the STSConnection object and pass the role ARN and a role session name.
assumed_role_object = sts_client.assume_role(
RoleArn=os.environ['ROLE_ARN'],
RoleSessionName="AssumeRoleFromCloudOperations"
)
# From the response that contains the assumed role, get the temporary credentials
credentials = assumed_role_object['Credentials']
assumed_role_session = boto3.Session(
aws_access_key_id=credentials['AccessKeyId'],
aws_secret_access_key=credentials['SecretAccessKey'],
aws_session_token=credentials['SessionToken']
)
# Check the queue size
def sqs():
queue_size = sqs_client.get_queue_attributes(
QueueUrl=os.environ['SQS_QUEUE_URL'],
AttributeNames=['ApproximateNumberOfMessages']
)
messages = int(queue_size["Attributes"]["ApproximateNumberOfMessages"])
return messages
# Count the number of active ec2 instances
def count_instances(ec2):
total_instances = 0
instances = ec2.instances.filter(Filters=[
{
'Instance State': 'instance-state-name',
'Values': ['running'],
'Name': 'tag:Name',
'Values': ['NameOfInstance']
},
])
for _ in instances:
total_instances += 1
return total_instances
print(f"Total number of active scan servers is: {total_instances}")
# Define the SNS Topic which will be integrated with OpsGenie
def sns():
topicArn = os.environ['SNS_ARN']
# Evaluate the set of conditions
def evaluate_conditions(context, event):
sqs()
if messages > int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances > int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS']):
print('False alert')
logger.info()
elif messages < int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances < int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS']):
print('False alert')
logger.info()
elif messages < int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances > int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS']):
print('False alert')
logger.info()
else:
sns.publish(TopicArn=os.environ['SNS_ARN'],
Message='sameple message',
Subject='sample subject')
print("Published to SNS Topic")
处理程序是 handler.evaluate_conditions
我的问题是我怎样才能在这个 lambda 函数中有一些结构? 当我 运行 函数出现命名错误时:
{
"errorMessage": "name 'messages' is not defined",
"errorType": "NameError",
"stackTrace": [
" File \"/var/task/mdc_alert/handler.py\", line 67, in evaluate_conditions\n if messages > int(os.environ['AVG_QUEUE_SIZE']) and count_instances.total_instances > int(\n"
]
}
所以我似乎无法在 evaluate_conditions() 函数中使用消息变量。 如何使“消息”和“total_instances”变量在 evaluate_conditions() 函数中可用?
我完全基于 google 搜索、Whosebug 和 boto3 文档编写了这个函数,因为我没有任何编程经验。
这个结构好不好,还是需要彻底改造? 我是否需要更改函数的顺序,或者创建一个 class?
眼前的问题是 messages
变量未定义。您的 sqs
函数 return 是一个值,但由于您是在 void 上下文中调用它,因此您实际上并没有对该值执行任何操作。您可以通过更改此行来解决此问题:
sqs()
给这个:
messages = sqs()
我还发现 count_instances 函数存在一些问题。它期望收到一个 ec2
变量,但您从 evaluate_conditions 错误地调用了它。您可以将 ec2_client 变量传递给它,也可以直接从函数内部使用 ec2_client 变量。
我建议重命名您的函数以更准确地反映它们的 return 值:
sqs -> sqs_msg_count
count_instances -> running_ec2_count
进行这些更改将使您能够重构 evaluate_conditions 以缩短 if-then 行,使您的代码整体上更易于阅读和遵循。如果您考虑了所有这些建议,您的代码可能如下所示:
# Check the queue size
def sqs_msg_count():
messages = sqs_client.get_queue_attributes(
QueueUrl=os.environ['SQS_QUEUE_URL'],
AttributeNames=['ApproximateNumberOfMessages']
)
return int(messages["Attributes"]["ApproximateNumberOfMessages"])
# Count the number of active ec2 instances
def running_instance_count():
running_instances = ec2_client.instances.filter(Filters=[
{
'Instance State': 'instance-state-name',
'Values': ['running'],
'Name': 'tag:Name',
'Values': ['NameOfInstance']
},
])
return len(running_instances)
# Evaluate the set of conditions
def evaluate_conditions(context, event):
sqs_count = sqs_msg_count()
sqs_average = int(os.environ['AVG_QUEUE_SIZE'])
ec2_count = running_instance_count()
ec2_average = int(os.environ['AVG_NR_OF_EC2_SCAN_SERVERS'])
if sqs_count > sqs_average and ec2_count > ec2_average:
print('False alert')
logger.info()
elif sqs_count < sqs_average and ec2_count < ec2_average:
print('False alert')
logger.info()
elif sqs_count < sqs_average and ec2_count > ec2_average:
print('False alert')
logger.info()
else:
sns_client.publish(
TopicArn=os.environ['SNS_ARN'],
Message='sameple message',
Subject='sample subject'
)
print("Published to SNS Topic")