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.
It's currently fetching forms and approving them. I want to add another feature where it would update the status of manually created projects.
Forms and Projects are different things.
You can share your code using AWS Lambda Layers. For example define them using AWS::Lambda::LayerVersion or AWS::Serverless::LayerVersion. You can then reference to them in your Python Lambda functions.
Here using AWS SAM :
MyLambdaFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: function_code/
Handler: app.lambda_handler
Runtime: python3.6
Layers:
- !Ref MySharedLayer
MySharedLayer:
Type: AWS::Serverless::LayerVersion
Properties:
LayerName: SharedLayerName
Description: Some shared code
ContentUri: layer_code/
CompatibleRuntimes:
- python3.6
RetentionPolicy: Retain
Each Lambda function will have the shared code available in /opt. It can be then used in the functions.
Now that the Lambda Layers are released you can easily share libraries and code between your Lambda Functions.
You can create a zip file for the Layer pretty much the same way as you can do so for a Function.
To share pymysql package you will need to create a Lambda Layer on base of the following function:
pymysql-bundle.zip/
python/lib/python3.7/site-packages/pymysql
Then from your Lambda Function's code you can reference it like this:
from pymysql import ...
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.
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)
You can call the other functions from the handler function, such as:
def list_users():
# list iam users here
def list_user_tags():
# list tags of these users here
def something_else():
# do something else
def handler_name(event, context):
users = list_users()
tags = list_user_tags(users)
return something_else(tags)
You can define a service class that is separate from the handler. Based on the event, you can invoke different methods in service class. You can also extend this idea more via Factory pattern.
lambda(event) {
...inspect event
...instantiate service class
...use service class methods to achieve goal
}
There are several different answers I can give here, from your specific question to more general concerns. So from most specific to most general:
Q. Can you put multiple statements in a lambda?
A. No. But you don't actually need to use a lambda. You can put the statements in a def instead. i.e.:
def second_lowest(l):
l.sort()
return l[1]
map(second_lowest, lst)
Q. Can you get the second lowest item from a lambda by sorting the list?
A. Yes. As alex's answer points out, sorted() is a version of sort that creates a new list, rather than sorting in-place, and can be chained. Note that this is probably what you should be using - it's bad practice for your map to have side effects on the original list.
Q. How should I get the second lowest item from each list in a sequence of lists?
A. sorted(l)[1] is not actually the best way for this. It has O(N log(N)) complexity, while an O(n) solution exists. This can be found in the heapq module.
>>> import heapq
>>> l = [5,2,6,8,3,5]
>>> heapq.nsmallest(l, 2)
[2, 3]
So just use:
map(lambda x: heapq.nsmallest(x,2)[1], list_of_lists)
It's also usually considered clearer to use a list comprehension, which avoids the lambda altogether:
[heapq.nsmallest(x,2)[1] for x in list_of_lists]
Putting the expressions in a list may simulate multiple expressions:
E.g.:
lambda x: [f1(x), f2(x), f3(x), x+1]
This will not work with statements.
When using AWS-Lambda, there is a difference between the way you implement your code, and the APIs your lambda exposes to the rest of the world.
In your case, it doesn't matter if the functions are of the same class or in different modules, the thing that matters is that you have to link each function call to the correct api-gateway. i would recommend reading about serverless infrastructure , a very useful tool for managing serverless applications, specifically this page of the docs describing function configuration.
Hope this helps!
You can have multiple functions in a single class. It's just that you have to set the required function as a handler for a particular API gateway on AWS which you are using it for the lambda function that you created. Then when we run our lambda it will only call the set handler function leaving the rest of the functions untouched.
You can even use rest of the function to work with other lambdas or even as a normal function which is called by the handler function internally.
You need to go through the aws lambda documentation for more information.
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.