关于python 代码结构、函数和调用函数的问题

Question

我试图创建一个函数来执行以下操作。（不知道是否值得一提，但这个函数是由另一个函数调用的）

使用 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?

Answer 1

眼前的问题是 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")

关于python 代码结构、函数和调用函数的问题

Question about python code structure, functions and calling functions

python-3.x

function

lambda

code-structure