Python 执行 aws lambda 时出现 ImportError

Question

我在 python 中编写了 AWS lambda 以向 sns 主题发送消息。

我正在使用 aws 代码管道在云中部署此代码。

运行这个 lambda 通过调用 api 网关。

python代码如下：

import boto3

from shared.environment import Env
from account.initiate_transition.transition_event_service import TransitionEventService
from shared.utils import Logger
from shared.xray_utils import trace
from shared.keep_warm import keep_warm


LOG = Logger.initialise_logger('transition-sms', None)
ENV = Env({Env.SMS_SNS_TOPIC_ARN, Env.AWS_REGION_NAME, Env.SMS_TRANSITION_TABLE})
SNS_CLIENT = boto3.client('sns')
DYNAMO_DB = boto3.resource('dynamodb', region_name=ENV.get(Env.AWS_REGION_NAME))
COGNITO_CLIENT = boto3.client('cognito-idp', region_name=ENV.get(Env.AWS_REGION_NAME))
SMS_TRANSITION_TABLE = DYNAMO_DB.Table(ENV.get(Env.SMS_TRANSITION_TABLE))
EVENT_SERVICE = TransitionEventService(SMS_TRANSITION_TABLE,
                                       SNS_CLIENT,
                                       ENV.get(Env.SMS_SNS_TOPIC_ARN),
                                       COGNITO_CLIENT)


@trace
@keep_warm
def lambda_handler(event, context):
    LOG = Logger.initialise_logger('transition-sms', context.aws_request_id)
    try:
        return EVENT_SERVICE.handle(event)
    except Exception as e:
        LOG.error(str(e))

transition_event_service.py

from account.initiate_transition.transition_sms_event import TransitionSmsEvent
from shared.utils import ApiGatewayResponse, Logger
from shared.xray_utils import trace
from http import HTTPStatus
from uuid import uuid4
from botocore.exceptions import ClientError
from jose import jwt

LOG = Logger.get_logger(__name__)


class TransitionEventService:
    def __init__(self, sms_transition_table, sns_client, topic_arn, cognito_client):
        LOG.debug('Initialising TransitionEventService')
        self.sms_transition_table = sms_transition_table
        self.sns_client = sns_client
        self.topic_arn = topic_arn
        self.cognito_client = cognito_client

    @trace
    def handle(self, event):
        try:
            event_object = self.instantiate_event(event)
        except Exception as e:
            LOG.error(e)
            return ApiGatewayResponse.init(HTTPStatus.BAD_REQUEST, {
                'error': 'Invalid request'
            })

        quid = str(uuid4())
        LOG.info('quid {}'.format(quid))

        LOG.debug('Storing SMS transition details')
        self.sms_transition_table.put_item(Item={
                'id_token': event_object.id_token,
                'landing_page': event_object.landing_page,
                'quid': quid
            })

        # Get phone number claim and verified claim
        LOG.debug('Decoding id_token to get unverified claims')
        claims = jwt.get_unverified_claims(event_object.id_token)
        user_pool_id = claims['UserPoolId']
        username = claims['Username']

        url = "account/verify-transition?quid=123&&username=xyz"
        response = self.cognito_client.admin_get_user(
                UserPoolId = user_pool_id,
                Username = username
            )
        phone_number = response['phone_number']

        LOG.debug('Sending Transition SMS')
        self.send_transition_sms(url=url, phone_number=phone_number)
        LOG.debug('SMS sent to {}'.format(phone_number))

        return ApiGatewayResponse.init(HTTPStatus.OK)

    def instantiate_event(self, event):
        return TransitionSmsEvent(event)

    def send_transition_sms(self, url: str, phone_number: str):
        try:
            LOG.debug('Publishing SMS url to SNS Topic:{}'.format(self.topic_arn))
            self.sns_client.publish(
                TopicArn=self.topic_arn,
                Message=json.dumps({
                    'default': json.dumps({
                        'url': url,
                        'phone_number': phone_number
                    })
                }),
                MessageStructure='json'
            )
        except ClientError as e:
            LOG.error(e)
            raise e

我在云监视日志中遇到以下错误：

谁能帮我解决这个问题。

Answer 1

您必须导入函数所需的库。为此，您必须创建一个虚拟环境并激活它：

virtualenv v-env
source v-env/bin/activate

然后安装不同的库

pip install library1
pip install library2
...

最后您停用虚拟环境并打包您的函数：

deactivate
zip -r function.zip .

更多信息：https://amzn.to/2oif6Wv

Answer 2

如果您有 requirement.txt 那么请验证以下条目是否存在

python-jose-cryptodome==1.3.2

Answer 3

如果您使用 requirement.txt，您必须包含所需的依赖项。因此，一旦您运行或通过代码管道部署代码，它就会下载依赖项并压缩到工件中，您可以通过 s3 将其用于 python lambda。

一旦您通过 python lambda 调用 api 网关备份，导入时就不会失败。

no module named jose

这表明您没有在 requirement.txt 文件中指定依赖项。

jose 的依赖性要求是

python-jose-cryptodome==1.3.*

Answer 4

检查您的 requirement.txt 文件中是否添加了 python-jose-cryptodome==1.3.2。如果不添加它，这是您收到此错误的主要原因。

Python 执行 aws lambda 时出现 ImportError

Python ImportError while executing aws lambda

python

python-import

amazon-web-services

importerror

amazon-cloudwatchlogs