How To Have More Than One Handler In Aws Lambda Function?

I have a very large python file that consists of multiple defined functions. If you're familiar with AWS Lambda, when you create a lambda function, you specify a handler, which is

Solution 1:

Your single handler function will need to be responsible for parsing the incoming event, and determining the appropriate route to take. For example, let's say your other functions are called helper1 and helper2. Your Lambda handler function will inspect the incoming event and then, based on one of the fields in the incoming event (ie. let's call it EventType), call either helper1 or helper2, passing in both the event and context objects.

defhandler_name(event, context):
  if event['EventType'] == 'helper1':
    helper1(event, context)
  elif event['EventType'] == 'helper2':
    helper2(event, context)

defhelper1(event, context):
  passdefhelper2(event, context):
  pass

This is only pseudo-code, and I haven't tested it myself, but it should get the concept across.

Solution 2:

Little late to the game but thought it wouldn't hurt to share. Best practices suggest that one separate the handler from the Lambda's core logic. Not only is it okay to add additional definitions, it can lead to more legible code and reduce waste--e.g. multiple API calls to S3. So, although it can get out of hand, I disagree with some of those critiques to your initial question. It's effective to use your handler as a logical interface to the additional functions that will accomplish your various work. In Data Architecture & Engineering land it's often less-costly and more efficient to work in this manner. Particularly if you are building out ETL pipelines, following service-oriented architectural patterns. Admittedly, I'm a bit of a Maverick and some may find this unruly/egregious but I've gone so far as to build classes into my Lambdas for various reasons--e.g. centralized, data-lake-ish S3 buckets that accommodate a variety of file types, reduce unnecessary requests, etc...--and I stand by it. Here's an example of one of my handler files from a CDK example project I put on the hub awhile back. Hopefully it'll give you some useful ideas, or at the very least not feel alone in wanting to beef up your Lambdas.

import requests
import json
from requests.exceptions import Timeout
from requests.exceptions import HTTPError
from botocore.exceptions import ClientError
from datetime import date
import csv
import os
import boto3
import logging

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

classAsteroids:
    """Client to NASA API and execution interface to branch data processing by file type.
    Notes:
        This class doesn't look like a normal class. It is a simple example of how one might
        workaround AWS Lambda's limitations of class use in handlers. It also allows for 
        better organization of code to simplify this example. If one planned to add
        other NASA endpoints or process larger amounts of Asteroid data for both .csv and .json formats,
        asteroids_json and asteroids_csv should be modularized and divided into separate lambdas
        where stepfunction orchestration is implemented for a more comprehensive workflow.
        However, for the sake of this demo I'm keeping it lean and easy.
    """defexecute(self, format):
        """Serves as Interface to assign class attributes and execute class methods
        Raises:
            Exception: If file format is not of .json or .csv file types.
        Notes:
            Have fun!
        """
        self.file_format=format
        self.today=date.today().strftime('%Y-%m-%d')
        # method call below used when Secrets Manager integrated. See get_secret.__doc__ for more.# self.api_key=get_secret('nasa_api_key')
        self.api_key=os.environ["NASA_KEY"]
        self.endpoint=f"https://api.nasa.gov/neo/rest/v1/feed?start_date={self.today}&end_date={self.today}&api_key={self.api_key}"
        self.response_object=self.nasa_client(self.endpoint)
        self.processed_response=self.process_asteroids(self.response_object)
        if self.file_format == "json":
            self.asteroids_json(self.processed_response)
        elif self.file_format == "csv":
            self.asteroids_csv(self.processed_response)
        else:
            raise Exception("FILE FORMAT NOT RECOGNIZED")
        self.write_to_s3()

    defnasa_client(self, endpoint):
        """Client component for API call to NASA endpoint.
        Args:
            endpoint (str): Parameterized url for API call.
        Raises:
            Timeout: If connection not made in 5s and/or data not retrieved in 15s.
            HTTPError & Exception: Self-explanatory
        Notes:
            See Cloudwatch logs for debugging.
        """try:
            response = requests.get(endpoint, timeout=(5, 15))
        except Timeout as timeout:
            print(f"NASA GET request timed out: {timeout}")
        except HTTPError as http_err:
            print(f"HTTP error occurred: {http_err}")
        except Exception as err:
            print(f'Other error occurred: {err}')
        else:
            return json.loads(response.content)

    defprocess_asteroids(self, payload):
        """Process old, and create new, data object with content from response.
        Args:
            payload (b'str'): Binary string of asteroid data to be processed.
        """
        near_earth_objects = payload["near_earth_objects"][f"{self.today}"]
        asteroids = []
        for neo in near_earth_objects:
            asteroid_object = {
                "id" : neo['id'],
                "name" : neo['name'],
                "hazard_potential" : neo['is_potentially_hazardous_asteroid'],
                "est_diameter_min_ft": neo['estimated_diameter']['feet']['estimated_diameter_min'],
                "est_diameter_max_ft": neo['estimated_diameter']['feet']['estimated_diameter_max'],
                "miss_distance_miles": [item['miss_distance']['miles'] for item in neo['close_approach_data']],
                "close_approach_exact_time": [item['close_approach_date_full'] for item in neo['close_approach_data']]
            }
            asteroids.append(asteroid_object)

        return asteroids

    defasteroids_json(self, payload):
        """Creates json object from payload content then writes to .json file.
        Args:
            payload (b'str'): Binary string of asteroid data to be processed.
        """
        json_file = open(f"/tmp/asteroids_{self.today}.json",'w')
        json_file.write(json.dumps(payload, indent=4))
        json_file.close()

    defasteroids_csv(self, payload):
        """Creates .csv object from payload content then writes to .csv file.
        """
        csv_file=open(f"/tmp/asteroids_{self.today}.csv",'w', newline='\n')
        fields=list(payload[0].keys())
        writer=csv.DictWriter(csv_file, fieldnames=fields)
        writer.writeheader()
        writer.writerows(payload)
        csv_file.close()

    defget_secret(self):
        """Gets secret from AWS Secrets Manager
        Notes:
            Have yet to integrate into the CDK. Leaving as example code.
        """
        secret_name = os.environ['TOKEN_SECRET_NAME']
        region_name = os.environ['REGION']
        session = boto3.session.Session()
        client = session.client(service_name='secretsmanager', region_name=region_name)
        try:
            get_secret_value_response = client.get_secret_value(SecretId=secret_name)
        except ClientError as e:
            raise e
        else:
            if'SecretString'in get_secret_value_response:
                secret = get_secret_value_response['SecretString']
            else:
                secret = b64decode(get_secret_value_response['SecretBinary'])
        return secret

    defwrite_to_s3(self):
        """Uploads both .json and .csv files to s3
        """
        s3 = boto3.client('s3')
        s3.upload_file(f"/tmp/asteroids_{self.today}.{self.file_format}", os.environ['S3_BUCKET'], f"asteroid_data/asteroids_{self.today}.{self.file_format}")


defhandler(event, context):
    """Instantiates class and triggers execution method.
    Args:
        event (dict): Lists a custom dict that determines interface control flow--i.e. `csv` or `json`.
        context (obj): Provides methods and properties that contain invocation, function and
            execution environment information. 
            *Not used herein.
    """
    asteroids = Asteroids()
    asteroids.execute(event)