In today's fast-paced digital world, real-time data exchange is the key to responsive applications. Webhooks have become indispensable for enabling instant communication between systems. Webhooks play a crucial role in enabling real-time communication between applications, allowing them to respond to events as they occur. In this article, we'll explore how to create robust webhooks using FastAPI and AWS Lambda. FastAPI, known for its simplicity and performance, will serve as our webhook receiver, while AWS Lambda handles event processing.We'll build an endpoint to trigger a webhook event and another to receive and process webhook events. By the end, you'll be equipped to build a scalable and efficient webhook system that can react to events from various sources, enhancing your application's real-time capabilities. Let's get started!
Prerequisites:
- Basic knowledge of Python, AWS Lambda, and API Gateway
- Python virtual environment set up
Step 1: Setting Up the Environment
Before we begin, ensure you've set up a Python virtual environment for your project. If you're unfamiliar with this process, you can find resources online to guide you through it tap this link to learn more.
Step 2: Creating the FastAPI Application
In your project directory, create a file named main.py
. This file will house our FastAPI code.
# Import necessary libraries
from fastapi import FastAPI, HTTPException, Request, BackgroundTasks
from pydantic import BaseModel
import requests
import uvicorn
# Create a FastAPI app instance
app = FastAPI()
# Define the Pydantic schema for the payload
class PaymentEvent(BaseModel):
amount: float
description: str
webhook_url: str
# Define a function to send webhook requests
def send_webhook_request(payload: dict):
try:
response = requests.post("https://api-gateway-url/path", json=payload)
response.raise_for_status() # Raise an exception for non-2xx response status codes
except requests.exceptions.RequestException as e:
print(f"Error sending webhook request: {e}")
# Define the endpoint to simulate a payment event
@app.post("/payment")
async def payment_event(payment_event: PaymentEvent, background_tasks: BackgroundTasks):
if not payment_event.amount:
raise HTTPException(status_code=400, detail="Amount cannot be empty")
amount = payment_event.amount
description = payment_event.description
webhook_url = payment_event.webhook_url
# Add the send_webhook_request function as a background task
background_tasks.add_task(send_webhook_request, {
"amount": amount,
"description": description,
"webhook_url": webhook_url
})
response_data = {
"status": "success",
"message": "Payment event simulated successfully",
"amount": amount,
"description": description,
"webhook_url": webhook_url
}
return response_data
# Define the endpoint to receive webhook events
@app.post("/webhook")
async def receive_webhook(request: Request):
data = await request.json()
return {"message": "Webhook received successfully", "payload": data}
# Run the FastAPI app
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=4040)
Step 3: Understanding the Code
In the code snippet above:
We create a FastAPI app instance (
app
) and import the required libraries.A Pydantic schema,
PaymentEvent
, is defined to structure the payload that users will send.The
send_webhook_request
function is created to send webhook requests to a specified URL. It is essential for triggering the webhook event.We define an endpoint (
/payment
) to simulate a payment event. This endpoint receives aPaymentEvent
payload, validates it, and then adds thesend_webhook_request
function as a background task.Another endpoint (
/webhook
) is defined to receive webhook events and process them.
Step 4: Running the FastAPI Application
To run the FastAPI application, execute the following command in your terminal:
python app.py
To facilitate communication between our Lambda function and the local server, we will employ ngrok, a powerful tool for tunneling requests. Follow these steps to set up ngrok:
Installation: Begin by downloading and installing ngrok from its official website, available at https://ngrok.com/download.
-
Running ngrok: Once installed, execute the following command in your terminal, replacing '4040' with the port number of your local server:
ngrok http 4040
This command will create a forwarding interface that efficiently redirects requests made to our ngrok URL to our local server. Your terminal should display information similar to the following:
Session Status: online Account: email@gmail.com (Plan: Free) Update: Update available (version 2.3.41, Ctrl-U to update) Version: 2.3.40 Region: United States (us) Web Interface: [http://127.0.0.1:4041](http://127.0.0.1:4041) Forwarding: [http://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app](http://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app) -> [http://localhost:4040](http://localhost:4040) Forwarding: [https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app](https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app) ->
This setup allows seamless redirection of requests from the ngrok URL to our local server. It's a crucial step in ensuring smooth communication between the two components of your system.
Step 5 - Verifying Ngrok Functionality
To confirm that Ngrok is functioning correctly and to access your FastAPI Swagger documentation via Ngrok, follow these steps. Typically, you would access the documentation locally at localhost:4040/docs
, but with Ngrok, you'll use a modified URL. In this example, we'll use the URL https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app/docs
. Replace this URL with your own forwarding URL.
Access Your FastAPI Swagger Docs: Open your web browser and enter the modified URL in the address bar. In this case, it should be
https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app/docs
. Press Enter to navigate to the Swagger documentation page.Verify Ngrok Functionality: Once the page loads, you should see the Swagger documentation for your FastAPI project. This confirms that Ngrok is successfully forwarding requests to your local server.
Below is an image of what the Swagger documentation page loaded with Ngrok might look like:
This visual confirmation assures you that Ngrok is effectively bridging the gap between your local environment and the internet, making your FastAPI documentation accessible via a public URL.
You can insert the image of your Swagger documentation page as mentioned in the content. Incorporate this refined content into your article to guide your readers in verifying the functionality of Ngrok and accessing their FastAPI Swagger documentation.
Step 6: Creating and Testing our Lambda Function Locally with SAM
AWS SAM (Serverless Application Model) is an open-source framework designed for building serverless applications. It streamlines the process of creating, testing, and deploying Lambda functions locally. To get started with AWS SAM, follow these steps:
Installation: Begin by installing the
sam-cli
for your specific operating system. Detailed instructions can be found in the official AWS SAM documentation at https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/install-sam-cli.html.-
Project Initialization: After installing
sam-cli
, navigate to the directory where yourmain.py
file resides. In this directory, run the following command:
sam init
During initialization, you will be prompted to choose a template. Opt for the
HelloWorldExample
template, which is a good starting point. When asked to select a runtime, ensure it matches the Python version installed on your local machine for testing purposes.Here's an example of the prompts you might encounter:
Which template source would you like to use? 1 - AWS Quick Start Templates 2 - Custom Template Location Choice: 1 Choose an AWS Quick Start application template 1 - Hello World Example 2 - Data processing 3 - Hello World Example With Powertools ... (other options) Template: 1 Use the most popular runtime and package type? (Python and zip) [y/N]: n Which runtime would you like to use? ... (list of runtimes) Runtime: 18 # Choose the appropriate Python version What package type would you like to use? 1 - Zip 2 - Image Package type: 1 Would you like to enable X-Ray tracing on the function(s) in your application? [y/N]: y (X-Ray may incur additional costs) Would you like to enable monitoring using CloudWatch Application Insights? [y/N]: y (AppInsights monitoring may incur additional costs) Project name [sam-app]: hooks
Project Structure: AWS SAM will create a folder for your functions, and in your case, it will be named 'hooks'. You can navigate to the 'hooks/hello_world/app.py' folder. This is where you will edit your Lambda function code.
Step 7 - Modifying the AWS SAM Configuration
To enhance your AWS SAM configuration, follow these steps:
-
Updating the Events Section: Navigate to the
template.yml
file and replace the existingEvents
section with the following code:
Events: HelloWorld: Type: Api Properties: Path: /hooks Method: post
This modification configures an API Gateway event source for your Lambda function, enabling it to respond to POST requests at the
/hooks
path. -
Modifying the API Gateway Endpoint: Next, adjust the API Gateway endpoint configuration as follows:
HelloWorldApi: Description: API Gateway endpoint URL for Prod stage for Hello World function Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hooks/"
You can customize the endpoint as needed, but this format provides a standardized endpoint for your Hello World function within the API Gateway.
Step 8 - Updating the Lambda Function Code
Now, let's enhance the functionality of your Lambda function by modifying the lambda_handler function in the hooks/app.py file. Replace your existing app.py code with the following:
import json
import requests
def lambda_handler(event, context):
# Parse the JSON body from the incoming event
body = json.loads(event['body'])
response = {} # Initialize an empty response dictionary
# Check if the 'amount' in the body is less than 100
if body['amount'] < 100:
response['event'] = "payment_failed" # Set the event to "payment_failed"
response['message'] = "An error occurred" # Set a corresponding error message
else:
response['event'] = "payment_verified" # Set the event to "payment_verified"
response['message'] = "The payment was verified" # Set a success message
try:
# Send a POST request to a webhook URL with the response data
print(body['webhook_url']) # Print the webhook URL for debugging (optional)
ip = requests.post(body['webhook_url'], json=response, headers={'Content-type': 'application/json'})
# Return a JSON response indicating success
return {
"statusCode": 200,
"body": json.dumps({
"message": "Action sent",
# "location": ip.text.replace("\n", "")
}),
}
except requests.RequestException as e:
# Handle any exceptions that occur during the POST request
raise e
Here's an explanation of the code:
The function receives two parameters:
event
andcontext
. In this context,event
represents the input data for the Lambda function, andcontext
provides information about the execution environment.The code starts by parsing the JSON data from the
event['body']
, assuming that the incoming request contains a JSON payload.It initializes an empty
response
dictionary to store the response data that will be sent back.The code checks the 'amount' field in the JSON body. If the amount is less than 100, it sets the
response['event']
to "payment_failed" and theresponse['message']
to "An error occurred." This indicates that the payment failed.If the amount is greater than or equal to 100, it sets
response['event']
to "payment_verified" andresponse['message']
to "The payment was verified." This indicates that the payment was successful.The code then attempts to send a POST request to a webhook URL specified in
body['webhook_url']
. It sends theresponse
data as a JSON payload in the request.If the POST request is successful (no exceptions are raised), it returns a JSON response with a 200 status code and a "message" field indicating that the action was sent.
If an exception (such as a network error) occurs during the POST request, it is caught in the
except
block, and the Lambda function raises the exception to handle it. This allows you to handle errors that may occur when communicating with the webhook.
This code essentially processes incoming payment data, checks the amount, sends a response to a webhook URL, and handles any potential errors that may occur during the process.
Certainly, I'll rephrase the instructions for you:
Step 9 - Testing and Deploying Your Lambda Function
Now, let's proceed with testing your Lambda function. Follow these steps to ensure everything is working as expected:
-
Prepare Event Data: Start by navigating to the
hooks/events/events.json
file. Replace the content ofevent.json
with the expected payload for your test case. In your case, it should look like this:
{ "body": { "message": "hello world", "amount": 90, "webhook_url": "https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app/webhook" }, "path": "/prod/hooks", "resourcePath": "/hooks", "httpMethod": "POST" }
Ensure that you use the URL provided by ngrok (
https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app/webhook
) as thewebhook_url
. This URL will forward requests made to it tohttp://localhost:4040/webhook
. -
Build the SAM Application: After updating the event data, build your AWS SAM application by running the following command:
sam build
This command will prepare your application for testing and deployment.
-
Test the Lambda Function Locally: To test your Lambda function locally, use the following command:
sam local invoke HelloWorldFunction --event events/event.json
You should expect to receive the following response in your terminal:
Invoking app.lambda_handler (python3.10) Local image is up-to-date Using local image: public.ecr.aws/lambda/python:3.10-rapid-x86_64. Mounting /path/to/your/webhook/hooks/.aws-sam/build/HelloWorldFunction as /var/task:ro,delegated, inside runtime container START RequestId: [Your_Request_ID] Version: $LATEST [Your_Webhook_URL] {"statusCode": 200, "body": "{\"message\": \"Action sent\"}"} END RequestId: [Your_Request_ID] REPORT RequestId: [Your_Request_ID] Init Duration: 0.06 ms Duration: 1683.21 ms Billed Duration: 1684 ms Memory Size: 128 MB Max Memory Used: 128 MB
This response confirms that your Lambda function executed successfully, and you received a response with a status code of 200.
-
Deploy Your Lambda Function: To deploy your Lambda function, use the following command with the
--guided
flag:
sam deploy --guided
This command will guide you through the deployment process, allowing you to configure various deployment settings.If you deployment is successful head over to the aws management console to view your lambda function, you should have something in the image below:
Step 10 - Testing Your Webhooks
The next step in your journey is to test your webhooks by loading your documentation page. This test allows you to confirm that your webhooks are functioning correctly. Follow these steps to perform the test:
-
Load the Documentation Page: Begin by accessing your swagger documentation page. Once there, you will have the opportunity to input your webhook payload. In this scenario, the objective is to make a request through the
/payment
endpoint.
-
Input Payload: Enter the following payload, which is tailored for your use case. It specifies an amount of 4000, a description of "payment_made," and a webhook URL:
{ "amount": 4000, "description": "payment_made", "webhook_url": "https://38fe-2c0f-2a80-c4-3700-6a51-91da-f16e-2918.ngrok-free.app/webhook" }
- Make the Request: Once you've entered the payload, initiate the request. If everything is set up correctly, you should receive a success response on the documentation page.
-
Check the Logs: To confirm that the webhook was received, inspect your server logs. Look specifically for a request made to the
POST /webhook
endpoint. You can observe the request in the logs, verifying that it was indeed received and processed.
Here are sample logs from your local server that illustrate the successful webhook process:
INFO: Started server process [300044]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:4040 (Press CTRL+C to quit)
INFO: 105.113.20.255:0 - "GET /docs HTTP/1.1" 200 OK
INFO: 105.113.20.255:0 - "GET /openapi.json HTTP/1.1" 200 OK
INFO: 105.113.20.255:0 - "POST /payment HTTP/1.1" 200 OK
INFO: 105.113.20.159:0 - "POST /webhook HTTP/1.1" 200 OK
This log excerpt confirms the successful webhook operation. The POST /webhook
request was received, processed, and responded to with a status code of 200.
This is how you build a webhook with aws lambda and fastapi, if you have any questions please comment,I'll respond.
Top comments (0)