AWS Лямбда Python Обработчик с классом

Question

Здравствуйте, я разработал python скрипт, который работает локально, однако я бы хотел сделать sh его лямбда AWS, у меня возникли некоторые проблемы, особенно с созданием обработчика в классе. Я выяснил, как избавиться от «ошибки обработчика» в лямбде, создав функцию-обработчик вне класса, но, к сожалению, он не запускает остальную часть моего кода. Моя цель - разместить функцию «lambda_handler» либо внутри моего класса, либо заставить функцию вызывать класс. Любой совет действительно ценится!

#!/usr/bin/python

import sys
import os
import json
import time
from datetime import datetime, timedelta


key = 'OKTA_AUTH'
### key = os.environ['OKTA_AUTH'] #####
outcome = 'outcome.result eq "FAILURE"'
event_type = 'eventType eq "application.provision.user.deactivate"'
app_id = 'target.id eq "SOME OKTA APP ID"'
all_params = f'{event_type} and {app_id} and {outcome}'
api_url = 'https://domain.okta.com/api/v1/logs'
slack_url = "SLACK URL"

last_hour_date_time = datetime.utcnow() - timedelta(days=10)
since = str(last_hour_date_time.strftime('%Y-%m-%dT%H:%M:%S.000Z'))
actor_list=[]
unique_list=[]


class Events:
    def lambda_handler(event, context):
        okta_auth()



    def okta_auth(self):
        event_list=[]
        url = api_url.format()
        params = {
            'filter': all_params,
            'since': since
        }
        response = requests.get(url, params=params,
                            headers={'Accept': 'application/json', 
                            'authorization': key})
        response_json = response.json()

        for event_data in response_json:
            events = event_data['outcome']['reason']
            event_list.append(events)
            actors = event_data['actor']['alternateId']
            actor_list.append(actors)
        unique_set = set(actor_list)
        unique_list.append(unique_set)
        if event_list != []:
            self.post_slack()
        else:
            sys.exit(0)

    def post_slack(self):
        url = slack_url.format()
        payload =  "{\"text\": \" Twillio Flex provisioing failure. Please check the following users %s \"}" % (unique_list)
        requests.post(url, headers={'Accept': 'application/json'}, data=payload)

### newly added code 



if __name__ == "__main__": 
    Events().lambda_handler()


### end
####ORIGINAL CODE USED TO BE

#if __name__ == "__main__":
#    Events().okta_auth()

Answer 1

после некоторого solid изучения я обнаружил, что столкнулся с двумя проблемами с моим кодом и тем, как работает AWS Lambda. Первым вопросом было то, как я вызывал класс в Lambda. Я подумал, что у вас должна быть функция внутри класса, но вместо этого я создал функцию для запуска класса.

def lambda_handler(event, context):
        Events().okta_auth() #### the function (okta_auth) within the class (Events)

Вторая проблема заключалась в развертывании с использованием встроенного кода. В Lambda по умолчанию не установлен модуль запросов. Поэтому я создал локальный каталог, в который затем установил запросы pip3, и переместил скрипт python в. Затем вы можете сжать содержимое папки и загрузить ее в aws lambda.

mkdir lambda_deploy
pip3 install --target ./lambda_deploy/ requests
cd lambda_deploy/
zip -r9 ${OLDPWD}/function.zip .

Вот окончательный код ниже для справки.

#!/usr/bin/python
import requests
import sys
import os
import json
import time
from datetime import datetime, timedelta

key = os.environ['OKTA_AUTH']
outcome = 'outcome.result eq "FAILURE"'
event_type = 'eventType eq "application.provision.user.deactivate"'
target_type = 'target.type eq "User"'
app_id = 'target.id eq "SOME APP ID"'
all_params = f'{event_type} and {target_type} and {app_id} and {outcome}'
api_url = f'https://domain.okta.com/api/v1/logs'
slack_url = "some slack WEBHOOK URL"
last_hour_date_time = datetime.utcnow() - timedelta(days=1)
since = str(last_hour_date_time.strftime('%Y-%m-%dT%H:%M:%S.000Z'))
unique_set=[]

def lambda_handler(event, context):
        Events().okta_auth()

class Events:
    def okta_auth(self):
        event_list=[]
        url = api_url.format()
        params = {
            'filter': all_params,
            'since': since
        }
        response = requests.get(url, params=params,
                            headers={'Accept': 'application/json', 
                            'authorization': key})
        response_json = response.json()
        for event_data in response_json:
            events = event_data['outcome']['reason']
            targets = event_data['target']
            parse = list(map(lambda x: x['alternateId'], targets))
            target_list=[]
            event_list.append(events)
            target_list.append(parse[1])
            for item in target_list:
                if item not in unique_set:
                    unique_set.append(item)
        if event_list != []:
            self.post_slack()
        else:
            print("no errors today")

    def post_slack(self):
        url = slack_url.format()
        payload =  "{\"text\": \" Twilio Flex provisioing failure. Please check the following users: \n %s \"}" % '\n'.join(unique_set)        
        requests.post(url, headers={'Accept': 'application/json'}, data=payload)

Answer 2

Ваша функция

def lambda_handler(event, context):
  print(event)

только печатает событие и больше ничего не выполняет. Наверное, поэтому лямбда ничего не делает. Lambda_handler - это точка входа вашей лямбды.