执行 aws lambda 时出现 Python ImportError

Question

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

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

通过调用 api gateway 来运行这个 lambda。

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

如果您有requirement.txt，请确认以下条目是否存在

python-jose-cryptodome==1.3.2 

Answer 2

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

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

然后你安装不同的库

pip install library1
pip install library2
...

最后你停用虚拟环境并打包你的功能：

deactivate
zip -r function.zip .

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

Answer 3

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

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

no module named jose

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

jose 的依赖要求是

python-jose-cryptodome==1.3.*

Answer 4

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