{
    "production": {
       ...
       "events": [{
            "function": "yourmodule.processupload_function",
            "event_source": {
                  "arn":  "arn:aws:s3:::my-bucket",
                  "events": [
                    "s3:ObjectCreated:*" // Supported event types: http://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html#supported-notification-event-types
                  ],
                  "key_filters": [{ // optional
                    "type": "suffix",
                    "value": "yourfile.json"
                  },
                  {
                    "type": "prefix",
                    "value": "prefix/for/your/object"
                  }]
               }
            }],
       ...
    }
}

And then:

$ zappa schedule production

And now your function will execute every time a new upload appears in your bucket!

To access the key's information in your application context, you'll want processuploadfunction to look something like this:

import boto3
s3_client = boto3.client('s3')
def processuploadfunction(event, context):     """     Process a file upload.     """
# Get the uploaded file's information     bucket = event['Records'][0]['s3']['bucket']['name'] # Will be my-bucket     key = event['Records'][0]['s3']['object']['key'] # Will be the file path of whatever file was uploaded.
# Get the bytes from S3     s3client.downloadfile(bucket, key, '/tmp/' + key) # Download this file to writable tmp space.     file_bytes = open('/tmp/' + key).read()

Similarly, for a Simple Notification Service event:

"events": [
            {
                "function": "yourmodule.yourfunction",
                "event_source": {
                    "arn":  "arn:aws:sns:::your-event-topic-arn",
                    "events": [
                        "sns:Publish"
                    ]
                }
            }
        ]

Optionally you can add SNS message filters:

"events": [
            {
                "function": "yourmodule.yourfunction",
                "event_source": {
                    "arn":  "arn:aws:sns:::your-event-topic-arn",
                    "filters": {
                        "interests": ["python", "aws", "zappa"],
                        "version": ["1.0"]
                    },
                    ...
                }
            }
        ]

DynamoDB and Kinesis are slightly different as it is not event based but pulling from a stream:

"events": [
           {
               "function": "replication.replicate_records",
               "event_source": {
                    "arn":  "arn:aws:dynamodb:us-east-1:1234554:table/YourTable/stream/2016-05-11T00:00:00.000",
                    "startingposition": "TRIMHORIZON", // Supported values: TRIM_HORIZON, LATEST
                    "batch_size": 50, // Max: 1000
                    "enabled": true // Default is false
               }
           }
       ]

SQS is also pulling messages from a stream. Read the AWS Documentation carefully since Lambda calls the SQS DeleteMessage API on your behalf once your function completes successfully. By default, if your function encounters an error while processing a batch, all messages in that batch become visible in the queue again.

"events": [
           {
               "function": "yourmodule.processmessages",
               "event_source": {
                    "arn":  "arn:aws:sqs:us-east-1:12341234:your-queue-name-arn",
                    "batch_size": 10, // Maximum: 10 for FIFO and 10,000 for Standard. Use 1 to trigger immediate processing
                    "enabled": true // Default is false
               }
           }
       ]

For configuring Lex Bot's intent triggered events:

"bot_events": [
        {
            "function": "lexbot.handlers.book_appointment.handler",
            "event_source": {
                "arn": "arn:aws:lex:us-east-1:01234123123:intent:TestLexEventNames:$LATEST", // optional. In future it will be used to configure the intent
                "intent":"intentName", // name of the bot event configured
                "invocation_source":"DialogCodeHook", // either FulfillmentCodeHook or DialogCodeHook
            }
        }
    ]

Events can also take keyword arguments:

"events": [
            {
                "function": "yourmodule.yourrecurring_function", // The function to execute
                "kwargs": {"key": "val", "key2": "val2"},  // Keyword arguments to pass. These are available in the event
                "expression": "rate(1 minute)" // When to execute it (in cron or rate format)
            }
       ]

To get the keyword arguments you will need to look inside the event dictionary:

def yourrecurringfunction(event, context):
    mykwargs = event.get("kwargs")  # dict of kwargs given in zappasettings file

You can find more example event sources here.

Asynchronous Task Execution

Zappa also now offers the ability to seamlessly execute functions asynchronously in a completely separate AWS Lambda instance!

For example, if you have a Flask API for ordering a pie, you can call your bake function seamlessly in a completely separate Lambda instance by using the zappa.asynchronous.task decorator like so:

from flask import Flask
from zappa.asynchronous import task
app = Flask(name)
@task def make_pie():     """ This takes a long time! """     ingredients = get_ingredients()     pie = bake(ingredients)     deliver(pie)
@app.route('/api/order/pie') def order_pie():     """ This returns immediately! """     make_pie()     return "Your pie is being made!"

And that's it! Your API response will return immediately, while the make_pie function executes in a completely different Lambda instance.

When calls to @task decorated functions or the zappa.asynchronous.run command occur outside of Lambda, such as your local dev environment, the functions will execute immediately and locally. The zappa asynchronous functionality only works when in the Lambda environment or when specifying Remote Invocations.

Catching Exceptions

Putting a try..except block on an asynchronous task like this:

@task
def make_pie():
    try:
        ingredients = get_ingredients()
        pie = bake(ingredients)
        deliver(pie)
    except Fault as error:
        """send an email"""
    ...
    return Response('Web services down', status=503)

will cause an email to be sent twice for the same error. See asynchronous retries at AWS. To work around this side-effect, and have the fault handler execute only once, change the return value to:

@task
def make_pie():
    try:
        """code block"""
    except Fault as error:
        """send an email"""
    ...
    return {} #or return True

Task Sources

By default, this feature uses direct AWS Lambda invocation. You can instead use AWS Simple Notification Service as the task event source by using the task_sns decorator, like so:

from zappa.asynchronous import task_sns
@task_sns

Using SNS also requires setting the following settings in your zappa_settings:

{
  "dev": {
    ..
      "async_source": "sns", // Source of async tasks. Defaults to "lambda"
      "async_resources": true, // Create the SNS topic to use. Defaults to true.
    ..
    }
}

This will automatically create and subscribe to the SNS topic the code will use when you call the zappa schedule command.

Using SNS will also return a message ID in case you need to track your invocations.

Direct Invocation

You can also use this functionality without a decorator by passing your function to zappa.asynchronous.run, like so:

from zappa.asynchronous import run
run(your_function, args, kwargs) # Using Lambda run(your_function, args, kwargs, service='sns') # Using SNS

Remote Invocations

By default, Zappa will use lambda's current function name and current AWS region. If you wish to invoke a lambda with a different function name/region or invoke your lambda from outside of lambda, you must specify the remoteawslambdafunctionname and remoteawsregion arguments so that the application knows which function and region to use. For example, if some part of our pizza making application had to live on an EC2 instance, but we wished to call the make_pie() function on its own Lambda instance, we would do it as follows:

@task(remoteawslambdafuncti, remoteaws_region='us-east-1')
def make_pie():
   """ This takes a long time! """
   ingredients = get_ingredients()
   pie = bake(ingredients)
   deliver(pie)

If those task() parameters were not used, then EC2 would execute the function locally. These same remoteawslambdafunctionname and remoteawsregion arguments can be used on the zappa.asynchronous.run() function as well.

Restrictions

The following restrictions to this feature apply:

• Functions must have a clean import path -- i.e. no closures, lambdas, or methods.
• args and kwargs must be JSON-serializable.
• The JSON-serialized arguments must be within the size limits for Lambda (256K) or SNS (256K) events.

Running Tasks in a VPC

If you're running Zappa in a Virtual Private Cloud (VPC), you'll need to configure your subnets to allow your lambda to communicate with services inside your VPC as well as the public Internet. A minimal setup requires two subnets.

In subnet-a:

• Create a NAT
• Create an Internet gateway
• In the route table, create a route pointing the Internet gateway to 0.0.0.0/0.

• Place your lambda function
• In the route table, create a route pointing the NAT that belongs to subnet-a to 0.0.0.0/0.

Some helpful resources are this tutorial, this other tutorial and this AWS doc page.

Responses

It is possible to capture the responses of Asynchronous tasks.

Zappa uses DynamoDB as the backend for these.

To capture responses, you must configure a asyncresponsetable in zappasettings. This is the DynamoDB table name. Then, when decorating with @task, pass captureresponse=True.

Async responses are assigned a response_id. This is returned as a property of the LambdaAsyncResponse (or SnsAsyncResponse) object that is returned by the @task decorator.

Example:

from zappa.asynchronous import task, getasyncresponse
from flask import Flask, abort, url_for, redirect, request, jsonify
from time import sleep
app = Flask(name)
@app.route('/payload') def payload():     delay = request.args.get('delay', 60)     x = longrunner(delay)     return redirect(urlfor('response', responseid=x.response_id))
@app.route('/async-response/<response_id>') def response(response_id):     response = getasyncresponse(response_id)     if response is None:         abort(404)
if response['status'] == 'complete':         return jsonify(response['response'])
sleep(5)
return "Not yet ready. Redirecting.", 302, {         'Content-Type': 'text/plain; charset=utf-8',         'Location': urlfor('response', responseid=response_id, backoff=5),         'X-redirect-reason': "Not yet ready.",     }
@task(capture_response=True) def longrunner(delay):     sleep(float(delay))     return {'MESSAGE': "It took {} seconds to generate this.".format(delay)}

Advanced Settings

There are other settings that you can define in your local settings to change Zappa's behavior. Use these at your own risk!

``javascript { "dev": { "additionaltextmimetypes": [], // allows you to provide additional mimetypes to be handled as text when binary_support is true. "albenabled": false, // enable provisioning of application load balancing resources. If set to true, you must fill out the albvpc_config option as well. "albvpcconfig": { "CertificateArn": "youracmcertificate_arn", // ACM certificate ARN for ALB "SubnetIds": [], // list of subnets for ALB "SecurityGroupIds": [] // list of security groups for ALB }, "apikeyrequired": false, // enable securing API Gateway endpoints with x-api-key header (default False) "apikey": "yourapikeyid", // optional, use an existing API key. The option "apikeyrequired" must be true to apply "app_type": "asgi", // optional, set to "asgi" to run an ASGI app (FastAPI, Starlette, Quart). When omitted, Zappa auto-detects async callables or defaults to WSGI. "apigateway_enabled": true, // Set to false if you don't want to create an API Gateway resource. Default true. "apigateway_description": "My funky application!", // Define a custom description for the API Gateway console. Default None. "assumepolicy": "myassume_policy.json", // optional, IAM assume policy JSON file "attachpolicy": "myattach_policy.json", // optional, IAM attach policy JSON file "apigatewaypolicy": "myapigateway_policy.json", // optional, API Gateway resource policy JSON file "functionurlenabled": false, // optional, set to true if you want to enable function URL. Default false. "functionurlconfig": { "authorizer": "NONE", // required if function url is enabled. default None. https://docs.aws.amazon.com/lambda/latest/dg/urls-auth.html "cors": { // set to false if disable cors. "allowedOrigins": [], // The origins that can access your function URL. default [*] "allowedHeaders": [], // The HTTP headers that origins can include in requests to your function URL. "allowedMethods": [], // The HTTP methods that are allowed when calling your function URL. For example: GET , POST , DELETE , or the wildcard character ( ). default [] "allowCredentials": false, //required, Whether to allow cookies or other credentials in requests to your function URL. default false. "exposedResponseHeaders": [], The HTTP headers in your function response that you want to expose to origins that call your function URL. "maxAge": 0 // The maximum amount of time, in seconds, that web browsers can cache results of a preflight request. default 0. } }, // NOTE: Function URLs do NOT include stage names in their paths. Unlike API Gateway v1/v2 which include // the stage name in the URL (e.g., /dev/mypath), Function URLs route directly to your app (e.g., /mypath). // This means SCRIPTNAME will be empty for Function URL requests, and PATHINFO will contain the full path. // // NOTE: When functionurlenabled is true with authorizer: "NONE", Zappa attaches TWO // resource-policy statements with Principal: "*": FunctionURLAllowPublicAccess for // lambda:InvokeFunctionUrl and FunctionURLAllowPublicAccessInvoke for lambda:InvokeFunction. // Both are required for unsigned calls to succeed; the AWS docs example at // https://docs.aws.amazon.com/lambda/latest/dg/urls-auth.html only shows the first statement, // but in practice the URL returns 403 AccessDeniedException without the second (see #1393). // If you are adding a Function URL manually (e.g. outside Zappa) and seeing 403s with NONE // auth, this two-statement shape is the missing piece. "apigateway_version": "v1", // optional, API Gateway version to use. Can be "v1" or "v2". Default "v1". "architecture": "x8664", // optional, Set Lambda Architecture, defaults to x8664. For Graviton 2 use: arm64 "async_source": "sns", // Source of async tasks. Defaults to "lambda" "async_resources": true, // Create the SNS topic and DynamoDB table to use. Defaults to true. "asyncresponsetable": "yourdynamodbtable_name", // the DynamoDB table name to use for captured async responses; defaults to None (can't capture) "asyncresponsetablereadcapacity": 1, // DynamoDB table read capacity; defaults to 1 "asyncresponsetablewritecapacity": 1, // DynamoDB table write capacity; defaults to 1 "awsendpointurls": { "awsservicename": "endpointurl" }, // a dictionary of endpointurls that emulate the appropriate service. Usually used for testing, for instance with localstack. "awsenvironmentvariables" : {"yourkey": "yourvalue"}, // A dictionary of environment variables that will be available to your deployed app via AWS Lambdas native environment variables. See also "environmentvariables" and "remoteenv" . Default {}. "awskmskeyarn": "yourawskmskey_arn", // Your AWS KMS Key ARN "aws_region": "aws-region-name", // optional, uses region set in profile or environment variables if not set here, "binary_support": true, // Enable automatic MIME-type based response encoding through API Gateway. Default true. "callbacks": { // Call custom functions during the local Zappa deployment/update process "settings": "myapp.settingscallback", // After loading the settings "zip": "myapp.zipcallback", // After creating the package "post": "myapp.postcallback", // After command has executed }, "cacheclusterenabled": false, // Use APIGW cache cluster (default False) "cacheclustersize": 0.5, // APIGW Cache Cluster size (default 0.5) "cacheclusterttl": 300, // APIGW Cache Cluster time-to-live (default 300) "cacheclusterencrypted": false, // Whether or not APIGW Cache Cluster encrypts data (default False) "certificate": "my_cert.crt", // SSL certificate file location. Used to manually certify a custom domain "certificatekey": "mykey.key", // SSL key file location. Used to manually certify a custom domain "certificatechain": "mycert_chain.pem", // SSL certificate chain file location. Used to manually certify a custom domain "certificate_arn": "arn:aws:acm:us-east-1:1234512345:certificate/aaaa-bbb-cccc-dddd", // ACM certificate ARN (needs to be in us-east-1 region). "cloudwatchloglevel": "OFF", // Enables/configures a level of logging for the given staging. Available options: "OFF", "INFO", "ERROR", default "OFF". "cloudwatchdatatrace": false, // Logs all data about received events. Default false. "cloudwatchmetricsenabled": false, // Additional metrics for the API Gateway. Default false. "cognito": { // for Cognito event triggers "user_pool": "user-pool-id", // User pool ID from AWS Cognito "triggers": [{ "source": "PreSignUp_SignUp", // triggerSource from http://docs.aws.amazon.com/cognito/latest/developerguide/cognito-user-identity-pools-working-with-aws-lambda-triggers.html#cognito-user-pools-lambda-trigger-syntax-pre-signup "function": "myapp.presignup_function" }] }, "contextheadermappings": { "HTTPheadername": "APIGatewaycontext_variable" }, // A dictionary mapping HTTP header names to API Gateway context variables "cors": false, // Enable Cross-Origin Resource Sharing. Default false. If true, simulates the "Enable CORS" button on the API Gateway console. Can also be a dictionary specifying lists of "allowedheaders", "allowedmethods", and string of "allowed_origin" "deadletterarn": "arn:aws:<sns/sqs>:::my-topic/queue", // Optional Dead Letter configuration for when Lambda async invoke fails thrice "debug": true, // Print Zappa configuration errors tracebacks in the 500. Default true. "deletelocalzip": true, // Delete the local zip archive after code updates. Default true. "deletes3zip": true, // Delete the s3 zip archive. Default true. "djangosettings": "yourproject.production_settings", // The modular path to your Django project's settings. For Django projects only. "domain": "yourapp.yourdomain.com", // Required if you're using a domain "base_path": "your-base-path", // Optional base path for API gateway custom domain base path mapping. Default None. Not supported for use with Application Load Balancer event sources. "environmentvariables": {"yourkey": "yourvalue"}, // A dictionary of environment variables that will be available to your deployed app. See also "remoteenv" and "awsenvironmentvariables". Default {}. "events": [ { // Recurring events "function": "yourmodule.yourrecurringfunction", // The function to execute (Pattern: [.A-Za-z0-9]+). "expression": "rate(1 minute)" // When to execute it (in cron or rate format) }, { // AWS Reactive events "function": "yourmodule.yourreactivefunction", // The function to execute (Pattern: [.A-Za-z0-9]+). "event_source": { "arn": "arn:aws:s3:::my-bucket", // The ARN of this event source "events": [ "s3:ObjectCreated:*" // The specific event to execute in response to. ] } } ], "endpoint_configuration": ["EDGE", "REGIONAL", "PRIVATE"], // Specify APIGateway endpoint None (default) or list ED