AWS Lambda Python Handler with class

Question

Hello I designed a python script that works locally however I would like to push it to AWS Lambda, I'm having some issues specifically with creating the handler within a class. I have figured out how to get rid of the 'handler error' in lambda by creating the handler function outside of the class but unfortunately that doesn't run the rest of my code. My goal is to place the "lambda_handler" function either inside my class or have the function call the class. Any advice is really appreciated!

#!/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()

JeremyM4n · Accepted Answer · 2022-08-21 14:34:27Z

After some solid studying, I discovered I was running into two issues with my code and how AWS Lambda works. The first issue was how I was calling the class in Lambda. I though that you had to have the function inside the class, but instead I created a function to run the class.

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

My second issue was deployment via inline code. Lambda does not have the requests module installed by default, so I created a local directory, where I then pip3 installed requests, and moved the python script to. You can then zip the folder contents and upload to aws lambda.

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

heres the final code below for reference.

#!/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)

Ghonima · Accepted Answer · 2020-02-08 10:26:02Z

-1

Your function

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

only prints the event and does not execute anything else. I guess that's why the lambda is not doing anything. The lambda_handler is the entry point of your lambda.

answered Feb 8, 2020 at 10:26

Ghonima

3,1122 gold badges17 silver badges25 bronze badges

1 Comment

Matthew Morcaldi Over a year ago

is it possible to add the lambda_handler function within my class and call the other functions while still having the Lambda_handler be recognized by lambda?

Collectives™ on Stack Overflow

AWS Lambda Python Handler with class

2 Answers 2

Comments

1 Comment

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Related