Question posted in
The official Twillio API documentation can be found here.

AWS Lambda Python Handler with class – Twillio

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()
Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    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)

  2. 0

    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.

    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search