AWS Lambda & API Gateway support for ASGI
Project description
Mangum
Mangum is an adapter for using ASGI applications with AWS Lambda & API Gateway. It is intended to provide an easy-to-use, configurable wrapper for any ASGI application deployed in an AWS Lambda function to handle API Gateway requests and responses.
Features
-
Multiple storage backend interfaces for managing WebSocket connections.
-
Compatibility with ASGI application frameworks, such as Starlette, FastAPI, and Quart.
-
Support for binary media types and payload compression in API Gateway.
-
Works with existing deployment and configuration tools, including Serverless Framework and AWS SAM.
-
Startup and shutdown lifespan events.
Requirements
Python 3.6+
Installation
pip install mangum
You can install the required dependencies for the WebSocket backends with one the following:
pip install mangum[aws]
pip install mangum[postgresql]
pip install mangum[redis]
Usage
The Mangum adapter class is designed to wrap any ASGI application, accepting various configuration options, returning a callable. It can wrap an application and be assigned to the handler:
from mangum import Mangum
# Define an ASGI application
handler = Mangum(app)
However, this is just one convention, you may also intercept events and construct the adapter instance separately:
def handler(event, context):
if event.get("some-key"):
# Do something or return, etc.
asgi_handler = Mangum(app)
response = asgi_handler(event, context) # Call the instance with the event arguments
return response
Examples
The examples below are "raw" ASGI applications with minimal configurations. You are more likely than not going to be using a framework, but you should be able to replace the app in these example with most ASGI framework applications. Please read the HTTP and WebSocket docs for more detailed configuration information.
HTTP
from mangum import Mangum
async def app(scope, receive, send):
await send(
{
"type": "http.response.start",
"status": 200,
"headers": [[b"content-type", b"text/plain; charset=utf-8"]],
}
)
await send({"type": "http.response.body", "body": b"Hello, world!"})
handler = Mangum(app)
WebSocket
from mangum import Mangum
html = b"""
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = new WebSocket("%s");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
""" % os.environ.get("WEBSOCKET_URL", "ws://localhost:3000")
async def app(scope, receive, send):
assert scope["type"] in ("http", "websocket")
if scope["type"] == "http":
message = await receive()
if message["type"] == "http.request":
await send(
{
"type": "http.response.start",
"status": 200,
"headers": [[b"content-type", b"text/html; charset=utf-8"]],
}
)
await send({"type": "http.response.body", "body": html})
if scope["type"] == "websocket":
while True:
message = await receive()
if message["type"] == "websocket.connect":
await send({"type": "websocket.accept"})
if message["type"] == "websocket.receive":
text = f"Received message: {message['text']}"
await send({"type": "websocket.send", "text": text})
if message["type"] == "websocket.disconnect":
await send({"type": "websocket.close", "code": 1000})
handler = Mangum(
app,
ws_config={
"backend": "s3",
"params": {
"bucket": "<s3-bucket-to-store-connections>"
}
}
)
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
