### Setup Development Environment Source: https://github.com/kludex/mangum/blob/main/CONTRIBUTING.md Run the setup script to create a Python virtual environment and install development dependencies. Requires uv to be installed. ```shell ./scripts/setup ``` -------------------------------- ### Manual Virtual Environment Setup Source: https://github.com/kludex/mangum/blob/main/CONTRIBUTING.md Manually create a Python virtual environment and install requirements using pip. ```shell python -m venv venv . venv/bin/active pip install -r requirements.txt ``` -------------------------------- ### Install Mangum Source: https://github.com/kludex/mangum/blob/main/docs/index.md Install Mangum using pip. This is the first step to using the adapter. ```shell pip install mangum ``` -------------------------------- ### Basic ASGI App with Mangum Source: https://github.com/kludex/mangum/blob/main/docs/index.md Example of a simple ASGI application and how to wrap it with Mangum. The lifespan option is turned off. ```python 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, lifespan="off") ``` -------------------------------- ### Basic ASGI Application Example Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Demonstrates a generic ASGI application structure that can be wrapped by Mangum. The framework details are abstracted away, focusing on the ASGI interface. ```python import mangum.adapter import framework from mangum import Mangum app = framework.applications.Application() @app.route("/") def endpoint(request: framework.requests.Request) -> dict: return {"hi": "there"} handler = Mangum(app) ``` -------------------------------- ### Quart Application with Route Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md An example of a Quart application defining a simple route. This application is then wrapped by Mangum for AWS Lambda deployment. ```python from quart import Quart from mangum import Mangum app = Quart(__name__) @app.route("/hello") async def hello(): return "hello world!" handler = Mangum(app) ``` -------------------------------- ### FastAPI Application with Routes Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md An example of a FastAPI application with two routes: a root endpoint and an endpoint with path parameters and query string support. This application is then wrapped by Mangum. ```python from fastapi import FastAPI from mangum import Mangum app = FastAPI() @app.get("/") def read_root(): return {"Hello": "World"} @app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q} handler = Mangum(app) ``` -------------------------------- ### Create Mangum Adapter with FastAPI Source: https://context7.com/kludex/mangum/llms.txt Instantiates the Mangum adapter with a FastAPI application. This example shows minimal configuration and an explicit configuration with path stripping and custom MIME types. The handler is called by Lambda like a normal function. ```python from fastapi import FastAPI from starlette.requests import Request from mangum import Mangum from mangum.exceptions import ConfigurationError app = FastAPI(title="My API") @app.get("/") def root(): return {"message": "Hello from Lambda"} @app.get("/items/{item_id}") def get_item(item_id: int, q: str | None = None): return {"item_id": item_id, "q": q} @app.get("/aws-context") def show_aws(request: Request): # The raw Lambda event and context are available via ASGI scope return { "aws_request_id": request.scope["aws.context"].get("aws_request_id"), "function_name": request.scope["aws.event"].get("requestContext", {}).get("apiId"), } # Minimal config — lifespan auto-detected, all defaults handler = Mangum(app) # Explicit config — deployed behind /v1 stage, strip transfer-encoding, add YAML as text handler = Mangum( app, lifespan="on", api_gateway_base_path="/v1", text_mime_types=[ "text/", "application/json", "application/javascript", "application/xml", "application/vnd.api+json", "application/vnd.oai.openapi", "application/x-yaml", # custom addition ], exclude_headers=["transfer-encoding", "x-internal-token"], ) # The handler is called by Lambda exactly like a normal handler function: # response = handler(event, context) # Expected response shape (API Gateway REST): # { # "statusCode": 200, # "headers": {"content-type": "application/json"}, # "multiValueHeaders": {}, # "body": "{\"message\":\"Hello from Lambda\"}", # "isBase64Encoded": False # } ``` -------------------------------- ### Quart App with Mangum Source: https://context7.com/kludex/mangum/llms.txt Integrate a Quart application with Mangum for AWS Lambda deployment. Includes setup for startup and shutdown events. ```python from quart import Quart from mangum import Mangum app = Quart(__name__) @app.before_serving async def startup(): app.db = "connected" @app.after_serving async def shutdown(): app.db = None @app.route("/ping") async def ping(): return {"db": app.db} handler = Mangum(app) ``` -------------------------------- ### Implement Raw ASGI App with Mangum Source: https://context7.com/kludex/mangum/llms.txt Wraps a raw ASGI application with Mangum, disabling the lifespan protocol. This example demonstrates how to handle raw request bodies and send plain text responses. ```python from mangum import Mangum async def raw_asgi_app(scope, receive, send): body = b"" if scope["type"] == "http": # Read full request body while True: msg = await receive() body += msg.get("body", b"") if not msg.get("more_body", False): break 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"Received: " + body, "more_body": False, }) handler = Mangum(raw_asgi_app, lifespan="off") ``` -------------------------------- ### FastAPI App with Mangum Source: https://github.com/kludex/mangum/blob/main/docs/index.md Example of integrating a FastAPI application with Mangum. The lifespan option is turned off. ```python from fastapi import FastAPI from mangum import Mangum app = FastAPI() @app.get("/") def read_root(): return {"Hello": "World"} @app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q} handler = Mangum(app, lifespan="off") ``` -------------------------------- ### Channels ASGI App with Mangum Handler Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Adapts a Channels ASGI application for Mangum, including compatibility for ASGI version 2 using `asgiref.compatibility.guarantee_single_callable`. Ensure Django and Channels are installed. ```python # asgi.py import os import django from channels.routing import get_default_application from asgiref.compatibility import guarantee_single_callable from mangum import Mangum os.environ.setdefault("DJANGO_SETTINGS_MODULE", "project.settings") django.setup() application = get_default_application() wrapped_application = guarantee_single_callable(application) handler = Mangum(wrapped_application, lifespan="off") ``` -------------------------------- ### Sanic App with Mangum Handler Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Integrates a Sanic application with Mangum. Ensure Sanic is installed and configured for async operations. ```python from sanic import Sanic from sanic.response import json from mangum import Mangum app = Sanic() @app.route("/") async def test(request): return json({"hello": "world"}) handler = Mangum(app) ``` -------------------------------- ### APIGateway Handler Usage Source: https://context7.com/kludex/mangum/llms.txt This example demonstrates how to use the `APIGateway` handler for AWS API Gateway REST (v1) events. It shows how to instantiate the handler with an event and configuration, access scope and body, and build a Lambda response. ```python from mangum.handlers.api_gateway import APIGateway event = { "resource": "/{proxy+}", "path": "/api/users", "httpMethod": "POST", "headers": { "Content-Type": "application/json", "Host": "abc.execute-api.us-west-2.amazonaws.com", "X-Forwarded-For": "203.0.113.5", "X-Forwarded-Port": "443", "X-Forwarded-Proto": "https", }, "multiValueHeaders": { "Content-Type": ["application/json"], }, "queryStringParameters": {"dry_run": "true"}, "multiValueQueryStringParameters": {"dry_run": ["true"]}, "pathParameters": {"proxy": "users"}, "requestContext": { "resourcePath": "/{proxy+}", "httpMethod": "POST", "identity": {"sourceIp": "203.0.113.5"}, "apiId": "abc", "stage": "prod", }, "body": '{"name": "Alice"}', "isBase64Encoded": False, "stageVariables": None, } config = {"api_gateway_base_path": "/", "text_mime_types": ["text/", "application/json"], "exclude_headers": []} handler = APIGateway(event, {}, config) print(handler.scope["method"]) # "POST" print(handler.scope["path"]) # "/api/users" print(handler.scope["query_string"]) # b"dry_run=true" print(handler.body) # b'{"name": "Alice"}' # Build a response (normally done internally by Mangum.__call__) response = {"status": 201, "headers": [[b"content-type", b"application/json"]], "body": b'{"id": 1}'} lambda_response = handler(response) # { # "statusCode": 201, # "headers": {"content-type": "application/json"}, # "multiValueHeaders": {}, # "body": '{"id": 1}', # "isBase64Encoded": False # } ``` -------------------------------- ### Mangum Handler Auto-Detection Source: https://context7.com/kludex/mangum/llms.txt This example shows how to use Mangum's `infer` method to automatically detect the appropriate handler for different AWS event types. It demonstrates detection for both API Gateway REST and HTTP API v2 events. ```python from mangum import Mangum from mangum.handlers import ALB, APIGateway, HTTPGateway, LambdaAtEdge async def app(scope, receive, send): ... mgmt = Mangum(app, lifespan="off") # Detect handler for an API Gateway REST event rest_event = { "resource": "/", "path": "/", "httpMethod": "GET", "requestContext": {"identity": {"sourceIp": "1.2.3.4"}, "httpMethod": "GET"}, "headers": {"Host": "x.execute-api.us-east-1.amazonaws.com", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443"}, "multiValueHeaders": {}, "queryStringParameters": None, "multiValueQueryStringParameters": None, "body": None, "isBase64Encoded": False, "stageVariables": None, } detected = mgmt.infer(rest_event, {}) print(type(detected).__name__) # "APIGateway" # Detect handler for an HTTP API v2 event http_v2_event = { "version": "2.0", "routeKey": "$default", "rawPath": "/items/42", "rawQueryString": "q=test", "headers": {"host": "x.execute-api.us-east-1.amazonaws.com", "x-forwarded-proto": "https", "x-forwarded-port": "443"}, "requestContext": { "http": {"method": "GET", "path": "/items/42", "sourceIp": "1.2.3.4"}, "accountId": "123456789012", "apiId": "abc", }, "body": None, "isBase64Encoded": False, } detected = mgmt.infer(http_v2_event, {}) print(type(detected).__name__) # "HTTPGateway" ``` -------------------------------- ### Basic Lifespan Event Handling with FastAPI Source: https://github.com/kludex/mangum/blob/main/docs/lifespan.md Demonstrates setting up startup and shutdown events in a FastAPI application and integrating it with Mangum using lifespan='auto'. ```python from mangum import Mangum from fastapi import FastAPI app = FastAPI() @app.on_event("startup") async def startup_event(): pass @app.on_event("shutdown") async def shutdown_event(): pass @app.get("/") def read_root(): return {"Hello": "World"} handler = Mangum(app, lifespan="auto") ``` -------------------------------- ### LifespanCycle __enter__ Method Source: https://github.com/kludex/mangum/blob/main/docs/lifespan.md Shows the implementation of the __enter__ method for LifespanCycle, which creates a task for the event loop to run the application startup. ```python def __enter__(self) -> None: """ Runs the event loop for application startup. """ self.loop.create_task(self.run()) self.loop.run_until_complete(self.startup()) ``` -------------------------------- ### Instantiate Mangum Adapter Source: https://github.com/kludex/mangum/blob/main/docs/adapter.md Basic instantiation of the Mangum adapter with default settings. All arguments are optional. ```python handler = Mangum( app, lifespan="auto", api_gateway_base_path=None, custom_handlers=None, text_mime_types=None, ) ``` -------------------------------- ### Clone and Set Up Upstream Remote Source: https://github.com/kludex/mangum/blob/main/CONTRIBUTING.md Clone the forked repository and add the upstream remote to keep your local copy in sync with the original repository. ```shell git clone git@github.com:/mangum.git cd mangum git remote add upstream git://github.com/jordaneremieff/mangum.git git fetch upstream ``` -------------------------------- ### LifespanCycle as a Context Manager Source: https://github.com/kludex/mangum/blob/main/docs/lifespan.md Illustrates how LifespanCycle is used as a context manager within the adapter class, handling startup and shutdown events when lifespan support is enabled. ```python with ExitStack() as stack: # Ignore lifespan events entirely if the `lifespan` setting is `off`. if self.lifespan in ("auto", "on"): asgi_cycle: typing.ContextManager = LifespanCycle( self.app, self.lifespan ) stack.enter_context(asgi_cycle) ``` -------------------------------- ### Responder Application with Route Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Demonstrates a Responder application with a dynamic route. Note the `static_dir` and `templates_dir` are set to `None` to disable automatic directory creation, which is necessary for AWS Lambda's read-only file system. ```python from mangum import Mangum import responder app = responder.API(static_dir=None, templates_dir=None) @app.route("/{greeting}") async def greet_world(req, resp, *, greeting): resp.text = f"{greeting}, world!" handler = Mangum(app) ``` -------------------------------- ### Lint and Format Code Source: https://github.com/kludex/mangum/blob/main/CONTRIBUTING.md Run static type checking with mypy and code formatting with black. ```shell ./scripts/lint ``` -------------------------------- ### Django ASGI App with Mangum Handler Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Sets up a Django ASGI application to work with Mangum. This configuration is standard for Django projects and requires `django.core.asgi.get_asgi_application`. ```python # asgi.py import os from mangum import Mangum from django.core.asgi import get_asgi_application os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings") application = get_asgi_application() handler = Mangum(application, lifespan="off") ``` -------------------------------- ### Wrap Starlette Application with Mangum Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Shows how to wrap an existing Starlette application with the Mangum adapter to make it compatible with AWS Lambda. ```python handler = Mangum(app) ``` -------------------------------- ### Integrate Starlette with Mangum Source: https://context7.com/kludex/mangum/llms.txt Demonstrates how to integrate the Starlette ASGI framework with Mangum to create a Lambda handler. The `handler` variable is the entry point registered in the AWS Lambda function configuration. ```python # --- Starlette --- from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route from mangum import Mangum async def homepage(request): return JSONResponse({"hello": "world"}) app = Starlette(routes=[Route("/", homepage)]) handler = Mangum(app) # Lambda handler: module.handler ``` -------------------------------- ### Add Upstream Remote and Fetch Source: https://github.com/kludex/mangum/blob/main/docs/contributing.md After cloning, add the upstream remote to sync with the original repository and fetch its branches. ```shell cd mangum git remote add upstream git://github.com/jordaneremieff/mangum.git git fetch upstream ``` -------------------------------- ### Manage ASGI Lifespan Events with Mangum Source: https://context7.com/kludex/mangum/llms.txt Configure Mangum to manage ASGI lifespan events. `lifespan="on"` will raise a 500 error on startup failure. State set during startup is copied to the HTTP request scope. ```python from contextlib import asynccontextmanager from fastapi import FastAPI from mangum import Mangum db_pool = {} # simulated connection pool @asynccontextmanager async def lifespan(app: FastAPI): # Startup: initialise resources db_pool["conn"] = "connected" print("DB pool ready:", db_pool) yield # Shutdown: release resources db_pool.clear() print("DB pool closed") app = FastAPI(lifespan=lifespan) @app.get("/status") def status(): return {"db": db_pool.get("conn", "disconnected")} # lifespan="on" means startup failure → 500, no silent swallow handler = Mangum(app, lifespan="on") # --- Testing lifespan state passing --- from mangum.types import Scope, Receive, Send async def stateful_app(scope: Scope, receive: Receive, send: Send): if scope["type"] == "lifespan": while True: msg = await receive() if msg["type"] == "lifespan.startup": scope["state"]["config"] = {"max_retries": 3} await send({"type": "lifespan.startup.complete"}) elif msg["type"] == "lifespan.shutdown": await send({"type": "lifespan.shutdown.complete"}) return if scope["type"] == "http": cfg = scope["state"]["config"] await send({"type": "http.response.start", "status": 200, "headers": [[b"content-type", b"application/json"]]}) import json await send({"type": "http.response.body", "body": json.dumps(cfg).encode()}) handler = Mangum(stateful_app, lifespan="on") # When invoked, scope["state"]["config"] == {"max_retries": 3} is available in the HTTP handler ``` -------------------------------- ### Mangum Exception Handling Source: https://context7.com/kludex/mangum/llms.txt Illustrates how Mangum raises specific exceptions for configuration errors and lifespan management issues. `ConfigurationError` is for invalid constructor arguments, and `LifespanFailure` propagates application-reported startup/shutdown failures. ```python from mangum import Mangum from mangum.exceptions import ConfigurationError, LifespanFailure from mangum.types import Scope, Receive, Send async def placeholder_app(scope: Scope, receive: Receive, send: Send): ... # ConfigurationError: invalid lifespan value try: handler = Mangum(placeholder_app, lifespan="maybe") except ConfigurationError as e: print(e) # "Invalid argument supplied for `lifespan`. Choices are: auto|on|off" # LifespanFailure: application sends startup failed async def failing_app(scope: Scope, receive: Receive, send: Send): if scope["type"] == "lifespan": while True: msg = await receive() if msg["type"] == "lifespan.startup": await send({"type": "lifespan.startup.failed", "message": "DB unreachable"}) elif scope["type"] == "http": await send({"type": "http.response.start", "status": 200, "headers": []}) await send({"type": "http.response.body", "body": b""}) handler = Mangum(failing_app, lifespan="on") mock_event = { "resource": "/", "path": "/", "httpMethod": "GET", "headers": {"Host": "x.execute-api.us-east-1.amazonaws.com", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443"}, "multiValueHeaders": {}, "queryStringParameters": None, "multiValueQueryStringParameters": None, "requestContext": {"resourcePath": "/", "httpMethod": "GET", "identity": {"sourceIp": "1.2.3.4"}, "apiId": "x"}, "body": None, "isBase64Encoded": False, "stageVariables": None, } try: handler(mock_event, {}) except LifespanFailure as e: print(f"Startup failed: {e}") # "Startup failed: Lifespan startup failure. DB unreachable" ``` -------------------------------- ### Run Tests Source: https://github.com/kludex/mangum/blob/main/CONTRIBUTING.md Execute all test cases using PyTest and generate a code coverage report. ```shell ./scripts/test ``` -------------------------------- ### Django Channels (ASGI v2 -> v3 Shim) with Mangum Source: https://context7.com/kludex/mangum/llms.txt Adapt Django Channels applications for Mangum using asgiref's compatibility layer. Lifespan is set to 'off'. ```python import os import django from channels.routing import get_default_application from asgiref.compatibility import guarantee_single_callable from mangum import Mangum os.environ.setdefault("DJANGO_SETTINGS_MODULE", "project.settings") django.setup() application = get_default_application() wrapped_application = guarantee_single_callable(application) handler = Mangum(wrapped_application, lifespan="off") ``` -------------------------------- ### Mangum Constructor Source: https://context7.com/kludex/mangum/llms.txt The `Mangum` class constructor creates the Lambda handler. It takes the ASGI application and several optional configuration parameters to customize its behavior. ```APIDOC ## `Mangum(app, lifespan, api_gateway_base_path, custom_handlers, text_mime_types, exclude_headers)` — Main adapter constructor Creates the Lambda handler. `app` is any ASGI 3.0 callable. `lifespan` controls lifespan event support (`"auto"` / `"on"` / `"off"`, default `"auto"`). `api_gateway_base_path` strips a path prefix before routing (default `"/"`). `custom_handlers` injects additional event-source handlers ahead of the built-ins. `text_mime_types` replaces the list of MIME types that are returned as plain text rather than base64. `exclude_headers` lists response header names (case-insensitive) to strip from the Lambda response. ```python # lambda_function.py — complete working example from fastapi import FastAPI from starlette.requests import Request from mangum import Mangum from mangum.exceptions import ConfigurationError app = FastAPI(title="My API") @app.get("/") def root(): return {"message": "Hello from Lambda"} @app.get("/items/{item_id}") def get_item(item_id: int, q: str | None = None): return {"item_id": item_id, "q": q} @app.get("/aws-context") def show_aws(request: Request): # The raw Lambda event and context are available via ASGI scope return { "aws_request_id": request.scope["aws.context"].get("aws_request_id"), "function_name": request.scope["aws.event"].get("requestContext", {}).get("apiId"), } # Minimal config — lifespan auto-detected, all defaults handler = Mangum(app) # Explicit config — deployed behind /v1 stage, strip transfer-encoding, add YAML as text handler = Mangum( app, lifespan="on", api_gateway_base_path="/v1", text_mime_types=[ "text/", "application/json", "application/javascript", "application/xml", "application/vnd.api+json", "application/vnd.oai.openapi", "application/x-yaml", # custom addition ], exclude_headers=["transfer-encoding", "x-internal-token"], ) # The handler is called by Lambda exactly like a normal handler function: # response = handler(event, context) # Expected response shape (API Gateway REST): # { # "statusCode": 200, # "headers": {"content-type": "application/json"}, # "multiValueHeaders": {}, # "body": "{\"message\":\"Hello from Lambda\"}", # "isBase64Encoded": False # } ``` ``` -------------------------------- ### Basic FastAPI App with Mangum Handler Source: https://github.com/kludex/mangum/blob/main/docs/http.md This snippet shows a basic FastAPI application integrated with Mangum, including GZip compression middleware and a custom text MIME type for binary responses. ```python from fastapi import FastAPI from fastapi.middleware.gzip import GZipMiddleware from mangum import Mangum app = FastAPI() app.add_middleware(GZipMiddleware, minimum_size=1000) @app.get("/") async def main(): return "somebigcontent" handler = Mangum(app, TEXT_MIME_TYPES=["application/vnd.some.type"]) ``` -------------------------------- ### Django App with Mangum Source: https://context7.com/kludex/mangum/llms.txt Deploy a Django application using Mangum. Lifespan is set to 'off' for partial compatibility. ```python import os from django.core.asgi import get_asgi_application from mangum import Mangum os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings") application = get_asgi_application() handler = Mangum(application, lifespan="off") # Django lifespan partial; use "off" ``` -------------------------------- ### Simulate Lambda Invocation (API Gateway REST) Source: https://context7.com/kludex/mangum/llms.txt This snippet demonstrates how to simulate a local AWS Lambda invocation using an API Gateway REST event. It includes a mock event and context, and shows how to call the handler and print the response. ```python mock_event = { "resource": "/{proxy+}", "path": "/echo", "httpMethod": "POST", "headers": { "Content-Type": "application/json", "Host": "abc123.execute-api.us-east-1.amazonaws.com", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443", }, "multiValueHeaders": {}, "queryStringParameters": None, "multiValueQueryStringParameters": None, "pathParameters": {"proxy": "echo"}, "requestContext": { "resourcePath": "/{proxy+}", "httpMethod": "POST", "identity": {"sourceIp": "1.2.3.4"}, "apiId": "abc123", }, "body": '{"hello": "world"}', "isBase64Encoded": False, "stageVariables": None, } mock_context = {} response = handler(mock_event, mock_context) print(response) # { # "statusCode": 200, # "headers": {"content-type": "text/plain; charset=utf-8"}, # "multiValueHeaders": {}, # "body": 'Received: {"hello": "world"}', # "isBase64Encoded": False # } ``` -------------------------------- ### Clone Mangum Repository Source: https://github.com/kludex/mangum/blob/main/docs/contributing.md Clone the Mangum repository from GitHub to your local machine. Replace `` with your GitHub username. ```shell git clone git@github.com:/mangum.git ``` -------------------------------- ### Mangum Handler Invocation Source: https://context7.com/kludex/mangum/llms.txt The `__call__` method is the entry point for the AWS Lambda runtime. It processes the incoming event and context, manages the ASGI lifecycle, and returns the response in a format compatible with AWS. ```APIDOC ## `Mangum.__call__(event, context)` — Invoke the adapter as a Lambda handler Infers the correct event-source handler from the event dict, runs the optional lifespan cycle, drives the HTTP cycle, and returns the AWS-formatted response dict. This is the method called directly by the Lambda runtime. ```python from mangum import Mangum async def raw_asgi_app(scope, receive, send): body = b"" if scope["type"] == "http": # Read full request body while True: msg = await receive() body += msg.get("body", b"") if not msg.get("more_body", False): break 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"Received: " + body, "more_body": False, }) handler = Mangum(raw_asgi_app, lifespan="off") ``` ``` -------------------------------- ### Exclude Headers with Mangum Source: https://context7.com/kludex/mangum/llms.txt Demonstrates how to exclude specific headers from being passed to the ASGI application using Mangum's configuration. ```python async def header_app(scope, receive, send): await send({ "type": "http.response.start", "status": 200, "headers": [ [b"content-type", b"text/plain; charset=utf-8"], [b"x-custom-header", b"should-be-removed"], [b"x-keep", b"keep-me"], ], }) await send({"type": "http.response.body", "body": b"Hello"}) from mangum.handlers.utils import handle_exclude_headers config = {"exclude_headers": ["x-custom-header"], "text_mime_types": [], "api_gateway_base_path": "/"} result = handle_exclude_headers({"content-type": "text/plain", "x-custom-header": "val", "x-keep": "keep-me"}, config) print(result) # {"content-type": "text/plain", "x-keep": "keep-me"} ``` -------------------------------- ### Create AWS Lambda Handler with FastAPI Source: https://github.com/kludex/mangum/blob/main/docs/adapter.md Wraps a FastAPI application to create an AWS Lambda handler. This is the most common convention for integrating ASGI apps with Lambda. ```python from mangum import Mangum from fastapi import FastAPI app = FastAPI() @app.get("/") def read_root(): return {"Hello": "World"} @app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q} handler = Mangum(app) ``` -------------------------------- ### Handle Binary Responses with `text_mime_types` Source: https://context7.com/kludex/mangum/llms.txt Configure `text_mime_types` to control which MIME types are treated as text. Non-text responses are base64-encoded with `"isBase64Encoded": True`. ```python from fastapi import FastAPI, Response from fastapi.middleware.gzip import GZipMiddleware from mangum import Mangum app = FastAPI() app.add_middleware(GZipMiddleware, minimum_size=500) @app.get("/image", response_class=Response) def get_image(): # 1×1 red GIF gif = ( b"GIF87a\x01\x00\x01\x00\x80\x01\x00\xff\x00\x00\x00\x00\x00," b"\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;" ) return Response(content=gif, media_type="image/gif") @app.get("/data.yaml", response_class=Response) def get_yaml(): return Response(content=b"name: Alice\nage: 30\n", media_type="application/x-yaml") # Without adding YAML to text_mime_types, /data.yaml body will be base64-encoded handler_default = Mangum(app, lifespan="off") # With YAML added, it is returned as plain text handler_with_yaml = Mangum( app, lifespan="off", text_mime_types=[ "text/", "application/json", "application/javascript", "application/xml", "application/vnd.api+json", "application/vnd.oai.openapi", "application/x-yaml", # ← added ], ) # At runtime, text_mime_types can also be mutated directly: handler_default.config["text_mime_types"].append("application/x-yaml") ``` -------------------------------- ### Configure API Gateway Base Path in Mangum Source: https://context7.com/kludex/mangum/llms.txt Shows how to configure Mangum to strip a custom base path from incoming API Gateway requests, ensuring the ASGI app receives clean paths. This is useful when API Gateway stages are mapped to custom domains with base path mappings. ```python from fastapi import FastAPI from mangum import Mangum app = FastAPI(root_path="/v1") @app.get("/users") def list_users(): return [{"id": 1}] # API Gateway stage mapped to domain at /v1 # Incoming path: /v1/users → app sees: /users handler = Mangum(app, lifespan="off", api_gateway_base_path="/v1") # Test: manually invoke with a prefixed path import asyncio mock_event = { "resource": "/{proxy+}", "path": "/v1/users", "httpMethod": "GET", "headers": {"Host": "api.example.com", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443"}, "multiValueHeaders": {}, "queryStringParameters": None, "multiValueQueryStringParameters": None, "requestContext": {"resourcePath": "/{proxy+}", "httpMethod": "GET", "identity": {"sourceIp": "1.1.1.1"}, "apiId": "x"}, "body": None, "isBase64Encoded": False, "stageVariables": None, } response = handler(mock_event, {}) print(response["statusCode"]) # 200 print(response["body"]) # '[{"id":1}]' ``` -------------------------------- ### Custom AWS Lambda Handler with Mangum Source: https://github.com/kludex/mangum/blob/main/docs/adapter.md Demonstrates intercepting AWS Lambda events before passing them to the Mangum adapter. Useful for custom event handling logic. ```python def handler(event, context): if event.get("some-key"): # Do something or return, etc. return asgi_handler = Mangum(app) response = asgi_handler(event, context) # Call the instance with the event arguments return response ``` -------------------------------- ### Starlette Application Definition Source: https://github.com/kludex/mangum/blob/main/docs/asgi-frameworks.md Defines a Starlette application with a simple homepage route. This application can then be wrapped by Mangum. ```python from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route from mangum import Mangum async def homepage(request): return JSONResponse({'hello': 'world'}) routes = [ Route("/", endpoint=homepage) ] app = Starlette(debug=True, routes=routes) ``` -------------------------------- ### Access AWS Event and Context in FastAPI Source: https://github.com/kludex/mangum/blob/main/docs/adapter.md Shows how to retrieve the AWS Lambda event and context from the ASGI scope within a FastAPI application. The event and context are available under 'aws.event' and 'aws.context' keys. ```python from fastapi import FastAPI from mangum import Mangum from starlette.requests import Request app = FastAPI() @app.get("/") def hello(request: Request): return {"aws_event": request.scope["aws.event"]} handler = Mangum(app) ``` -------------------------------- ### ALB Handler for Application Load Balancer Events Source: https://context7.com/kludex/mangum/llms.txt Handles Lambda events forwarded by an ALB target group, supporting single-value and multi-value headers. Decodes and re-encodes query parameters to prevent double-encoding. ```python from mangum.handlers.alb import ALB event = { "requestContext": { "elb": { "targetGroupArn": "arn:aws:elasticloadbalancing:us-east-2:123456789012:targetgroup/my-fn/abc" } }, "httpMethod": "GET", "path": "/health", "queryStringParameters": {"check": "deep"}, "headers": { "host": "my-alb-1234.us-east-2.elb.amazonaws.com", "x-forwarded-for": "203.0.113.1", "x-forwarded-port": "80", "x-forwarded-proto": "http", }, "body": None, "isBase64Encoded": False, } config = {"api_gateway_base_path": "/", "text_mime_types": ["text/", "application/json"], "exclude_headers": []} handler = ALB(event, {}, config) print(handler.scope["scheme"]) # "http" print(handler.scope["path"]) # "/health" # Multi-value headers disabled → single dict response response = { "status": 200, "headers": [ [b"content-type", b"application/json"], [b"set-cookie", b"a=1; Secure"], [b"set-cookie", b"b=2; Secure"], ], "body": b'{"status": "ok"}', } lambda_response = handler(response) # { # "statusCode": 200, # "headers": { # "content-type": "application/json", # "set-cookie": "a=1; Secure", # "Set-cookie": "b=2; Secure" # case-mutated to preserve both # }, # "body": '{"status": "ok"}', # "isBase64Encoded": False # } ``` -------------------------------- ### Lambda@Edge Handler for CloudFront Events Source: https://context7.com/kludex/mangum/llms.txt Handles CloudFront viewer-request or origin-request events. Mangum normalizes CloudFront's nested header format to the flat ASGI bytes format. ```python from mangum.handlers.lambda_at_edge import LambdaAtEdge event = { "Records": [ { "cf": { "config": { "distributionDomainName": "d111111abcdef8.cloudfront.net", "distributionId": "EDFDVBD6EXAMPLE", "eventType": "viewer-request", "requestId": "4TyzHTaYWb1GX1qTfsHhEqnmBZHBgXFqzOKkHsNTX30=", }, "request": { "clientIp": "203.0.113.178", "method": "GET", "uri": "/images/banner.png", "querystring": "v=2", "headers": { "host": [{"key": "host", "value": "d111111abcdef8.cloudfront.net"}], "x-forwarded-port": [{"key": "x-forwarded-port", "value": "443"}], "x-forwarded-proto": [{"key": "x-forwarded-proto", "value": "https"}], "cloudfront-forwarded-proto": [{"key": "cloudfront-forwarded-proto", "value": "https"}], }, }, } } ] } config = {"api_gateway_base_path": "/", "text_mime_types": ["text/", "application/json"], "exclude_headers": []} handler = LambdaAtEdge(event, {}, config) print(handler.scope["path"]) # "/images/banner.png" print(handler.scope["query_string"]) # b"v=2" print(handler.scope["scheme"]) # "https" # LambdaAtEdge response format uses numeric status and nested header dicts response = { "status": 200, "headers": [[b"content-type", b"image/png"], [b"cache-control", b"max-age=86400"]], "body": b"\x89PNG...", } lambda_response = handler(response) # { # "status": 200, # "headers": { # "content-type": [{"key": "content-type", "value": "image/png"}], # "cache-control": [{"key": "cache-control", "value": "max-age=86400"}], # }, # "body": "iVBORw...", # base64-encoded because image/png is not a text MIME type # "isBase64Encoded": True # } ``` -------------------------------- ### LifespanCycle __exit__ Method Source: https://github.com/kludex/mangum/blob/main/docs/lifespan.md Details the __exit__ method of LifespanCycle, responsible for running the event loop to execute application shutdown functions. ```python def __exit__( self, exc_type: typing.Optional[typing.Type[BaseException]], exc_value: typing.Optional[BaseException], traceback: typing.Optional[types.TracebackType], ) -> None: """ Runs the event loop for application shutdown. """ self.loop.run_until_complete(self.shutdown()) ``` -------------------------------- ### Pull Upstream Changes Source: https://github.com/kludex/mangum/blob/main/CONTRIBUTING.md Fetch and pull changes from the upstream repository to keep your local fork updated. ```shell git pull upstream main ``` -------------------------------- ### Mangum.infer(event, context) Source: https://context7.com/kludex/mangum/llms.txt Automatically detects and instantiates the appropriate handler for a given AWS Lambda event and context. It iterates through registered handlers, prioritizing custom ones, then built-in handlers like ALB, HTTPGateway, APIGateway, and LambdaAtEdge, until a suitable handler is found. ```APIDOC ## `Mangum.infer(event, context)` — Handler auto-detection Iterates through registered handlers (custom first, then built-ins: `ALB`, `HTTPGateway`, `APIGateway`, `LambdaAtEdge`) calling each handler's `.infer()` classmethod until one returns `True`, then instantiates and returns it. Raises `RuntimeError` if no handler matches. ```python from mangum import Mangum from mangum.handlers import ALB, APIGateway, HTTPGateway, LambdaAtEdge async def app(scope, receive, send): ... mgmt = Mangum(app, lifespan="off") # Detect handler for an API Gateway REST event rest_event = { "resource": "/", "path": "/", "httpMethod": "GET", "requestContext": {"identity": {"sourceIp": "1.2.3.4"}, "httpMethod": "GET"}, "headers": {"Host": "x.execute-api.us-east-1.amazonaws.com", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443"}, "multiValueHeaders": {}, "queryStringParameters": None, "multiValueQueryStringParameters": None, "body": None, "isBase64Encoded": False, "stageVariables": None, } detected = mgmt.infer(rest_event, {}) print(type(detected).__name__) # "APIGateway" # Detect handler for an HTTP API v2 event http_v2_event = { "version": "2.0", "routeKey": "$default", "rawPath": "/items/42", "rawQueryString": "q=test", "headers": {"host": "x.execute-api.us-east-1.amazonaws.com", "x-forwarded-proto": "https", "x-forwarded-port": "443"}, "requestContext": { "http": {"method": "GET", "path": "/items/42", "sourceIp": "1.2.3.4"}, "accountId": "123456789012", "apiId": "abc", }, "body": None, "isBase64Encoded": False, } detected = mgmt.infer(http_v2_event, {}) print(type(detected).__name__) # "HTTPGateway" ``` ``` -------------------------------- ### APIGateway Handler Source: https://context7.com/kludex/mangum/llms.txt Handles events originating from AWS API Gateway REST APIs (v1). This handler is designed for events that include 'resource' and 'requestContext' fields. It supports various features such as multi-value headers, base64-encoded bodies, multi-value query string parameters, and the stripping of an `api_gateway_base_path`. ```APIDOC ## `APIGateway` handler — AWS API Gateway REST (v1) event source Handles events from API Gateway REST APIs where the event contains `"resource"` and `"requestContext"`. Supports both `headers` and `multiValueHeaders`, base64-encoded bodies, multi-value query string parameters, and `api_gateway_base_path` stripping. ```python from mangum.handlers.api_gateway import APIGateway event = { "resource": "/{proxy+}", "path": "/api/users", "httpMethod": "POST", "headers": { "Content-Type": "application/json", "Host": "abc.execute-api.us-west-2.amazonaws.com", "X-Forwarded-For": "203.0.113.5", "X-Forwarded-Port": "443", "X-Forwarded-Proto": "https", }, "multiValueHeaders": { "Content-Type": ["application/json"], }, "queryStringParameters": {"dry_run": "true"}, "multiValueQueryStringParameters": {"dry_run": ["true"]}, "pathParameters": {"proxy": "users"}, "requestContext": { "resourcePath": "/{proxy+}", "httpMethod": "POST", "identity": {"sourceIp": "203.0.113.5"}, "apiId": "abc", "stage": "prod", }, "body": '{"name": "Alice"}', "isBase64Encoded": False, "stageVariables": None, } config = {"api_gateway_base_path": "/", "text_mime_types": ["text/", "application/json"], "exclude_headers": []} handler = APIGateway(event, {}, config) print(handler.scope["method"]) # "POST" print(handler.scope["path"]) # "/api/users" print(handler.scope["query_string"]) # b"dry_run=true" print(handler.body) # b'{"name": "Alice"}' # Build a response (normally done internally by Mangum.__call__) response = {"status": 201, "headers": [[b"content-type", b"application/json"]], "body": b'{"id": 1}'} lambda_response = handler(response) # { # "statusCode": 201, # "headers": {"content-type": "application/json"}, # "multiValueHeaders": {}, # "body": '{"id": 1}', # "isBase64Encoded": False # } ``` ```