One option is to use a boto3 paginator.

Alternatively, you could use a loop rather than a recursive function.

It would be something like:

response = api_call()
<do stuff with response>
while response['NextToken']:
    response=api_call(NextToken=response['NextToken'])
    <do stuff with response>

You can probably avoid having to double-up the <do stuff> bit by improving on the while statement.

Answer from John Rotenstein on Stack Overflow
๐ŸŒ
Reddit
reddit.com โ€บ r/aws โ€บ tutorial: deploying multiple python lambda functions using a single docker image
r/aws on Reddit: Tutorial: Deploying multiple Python Lambda functions using a single Docker image
October 9, 2022 - Regarding the anti-pattern; I am by no means an AWS expert, but I do think this approach makes a lot of sense when you have many Lambdas belonging to a single project. For example, as an alternative I also tried spinning up my Lambda functions with PythonFunction, which is currently in Alpha.
๐ŸŒ
Serverless Land
serverlessland.com โ€บ content โ€บ service โ€บ lambda โ€บ guides โ€บ aws-lambda-operator-guide โ€บ runtimes-multiple
AWS Lambda Operator Guide | Multiple runtimes in single applications | Serverless Land
Each Lambda function can use only one runtime but you can use multiple runtimes across multiple functions. This enables you to choose the best runtime for the task of the function.
๐ŸŒ
Amazon Web Services
docs.aws.amazon.com โ€บ aws lambda โ€บ developer guide โ€บ aws lambda functions โ€บ programming languages โ€บ building lambda functions with python
Building Lambda functions with Python - AWS Lambda
Your code runs in an environment ... from an AWS Identity and Access Management (IAM) role that you manage. To learn more about the SDK versions included with the Python runtimes, see Runtime-included SDK versions. Lambda supports the following Python runtimes. ... Choose Create function...
Find elsewhere
Top answer
1 of 2
22

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.

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

def helper1(event, context):
  pass

def helper2(event, context):
  pass

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

2 of 2
11

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)

class Asteroids:
    """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.
    """

    def execute(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()

    def nasa_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)

    def process_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

    def asteroids_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()

    def asteroids_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()

    def get_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

    def write_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}")


def handler(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)
๐ŸŒ
Amazon Web Services
docs.aws.amazon.com โ€บ aws lambda โ€บ developer guide โ€บ aws lambda functions โ€บ lambda runtimes
Lambda runtimes - AWS Lambda
November 18, 2021 - For functions with more complex ... in the Lambda handler. Choice of runtime is also influenced by developer preference and language familiarity. Each major programming language release has a separate runtime, with a unique runtime identifier, such as nodejs24.x or python3.14. To configure a function to use a new major language version, you need to change the runtime identifier. Since AWS Lambda cannot ...
๐ŸŒ
AWS
aws.amazon.com โ€บ blogs โ€บ compute โ€บ parallel-processing-in-python-with-aws-lambda
Parallel Processing in Python with AWS Lambda | Amazon Web Services
September 11, 2017 - If you develop an AWS Lambda function with Node.js, you can call multiple web services without waiting for a response due to its asynchronous nature. All requests are initiated almost in parallel, so you can get results much faster than a series of sequential calls to each web service. Considering the maximum execution duration for Lambda, it is beneficial for I/O bound tasks to run in parallel. If you develop a Lambda function with Python...
๐ŸŒ
AWS re:Post
repost.aws โ€บ questions โ€บ QUZacMmnGNRc-tJ7Euozn6PQ โ€บ multiple-runtimes-for-aws-lambda
Multiple Runtimes for AWS Lambda | AWS re:Post
May 23, 2024 - For example, in a Lambda function that transforms JSON between services, you could choose Node.js for your business logic. In another function handling data processing, you may choose Python. Both can operate in a single serverless application. For more info, please check https://docs.aws.amazon.com/lambda/latest/operatorguide/runtimes-multiple.html
๐ŸŒ
Medium
venkatesh29.medium.com โ€บ deploying-multiple-lambda-functions-under-a-single-lambda-project-b0bfd2f87f1e
Deploying Multiple Lambda Functions under a Single Lambda Project | by Venkatesh Gaddam | Medium
August 7, 2021 - In this Article I am going to explain how to write, test and deploy multiple Lambda Functions with the help of one single Lambda Project. Here, I will be creating three Lambda Functions to demonstrate about the topic. So, Letโ€™s Start Coding. Prerequisites needed to implement this are AWS Toolkit ...
๐ŸŒ
HackerNoon
hackernoon.com โ€บ aws-lambda-should-you-have-few-monolithic-functions-or-many-single-purposed-functions-8c3872d4338f
AWS Lambda โ€” should you have few monolithic functions or many single-purposed functions? | HackerNoon
January 8, 2018 - A funny moment (at 38:50) happened during Tim Brayโ€™s session (SRV306) at re:invent 2017, when he asked the audience if we should have many simple, single-purposed functions, or fewer monolithic functions, and the room was pretty much split in half.
๐ŸŒ
Symphonia
blog.symphonia.io โ€บ posts โ€บ 2022-07-20_lambda_event_routing
MonoLambdas, Nano Functions, and Goldilocks | The Symphonium
July 20, 2022 - In other words a larger code artifact ... multiple event types in one Lambda function (the MonoLambda style) then your code artifact must contain all of the code and libraries to support those events....
๐ŸŒ
Reddit
reddit.com โ€บ r/aws โ€บ multiple lambda functions from the same project?
r/aws on Reddit: Multiple lambda functions from the same project?
April 16, 2019 -

I am new to Lambda and I have started learning it only recently. When I upload a new Lambda function that I wrote in Java to AWS, it will package my project as a zip and uploads it to an S3 bucket. That's well and good. However, today I created a new handler class in the same project, and when I uploaded this to AWS, it became a new zip in the S3 bucket. So, now I have two zip files of the same project for each function, which I think is pretty redundant.

Is there any way for two AWS Lambda functions to reference the same project in S3? I think that perhaps there is some step that I have overlooked when uploading my functions.