How to use the mangum.Mangum function in mangum

To help you get started, we’ve selected a few mangum examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github erm / mangum / tests / test_lifespan.py View on Github external
nonlocal startup_complete
        startup_complete = True

    @app.after_serving
    async def on_shutdown():
        nonlocal shutdown_complete
        shutdown_complete = True

    @app.route(path)
    async def hello():
        return "hello world!"

    assert not startup_complete
    assert not shutdown_complete

    handler = Mangum(app)

    assert startup_complete
    assert not shutdown_complete

    response = handler(mock_http_event, {})

    assert response == {
        "statusCode": 200,
        "isBase64Encoded": False,
        "headers": {"content-length": "12", "content-type": "text/html; charset=utf-8"},
        "body": "hello world!",
    }
    assert startup_complete
    assert shutdown_complete
github erm / mangum / tests / test_websockets.py View on Github external
    @app.websocket_route("/ws")
    async def websocket_endpoint(websocket):
        await websocket.accept()
        await websocket.send_json({"url": str(websocket.url)})
        await websocket.close()

    handler = Mangum(app, enable_lifespan=False)
    response = handler(mock_ws_connect_event, {})
    assert response == {
        "body": "OK",
        "headers": {"content-type": "text/plain; charset=utf-8"},
        "isBase64Encoded": False,
        "statusCode": 200,
    }

    handler = Mangum(app, enable_lifespan=False)
    with mock.patch(
        "mangum.protocols.websockets.ASGIWebSocketCycle.send_data"
    ) as send_data:
        send_data.return_value = None
        response = handler(mock_ws_send_event, {})
        assert response == {
            "body": "OK",
            "headers": {"content-type": "text/plain; charset=utf-8"},
            "isBase64Encoded": False,
            "statusCode": 200,
        }
github erm / mangum / tests / test_http.py View on Github external
def test_http_cycle_state(mock_http_event) -> None:
    async def app(scope, receive, send):
        assert scope["type"] == "http"
        await send({"type": "http.response.body", "body": b"Hello, world!"})

    handler = Mangum(app, enable_lifespan=False)

    response = handler(mock_http_event, {})
    assert response == {
        "body": "Internal Server Error",
        "headers": {"content-type": "text/plain; charset=utf-8"},
        "isBase64Encoded": False,
        "statusCode": 500,
    }

    async def app(scope, receive, send):
        assert scope["type"] == "http"
        await send({"type": "http.response.start", "status": 200})
        await send({"type": "http.response.start", "status": 200})

    handler = Mangum(app, enable_lifespan=False)
github erm / mangum / tests / test_http.py View on Github external
"raw_path": None,
            "root_path": "",
            "scheme": "https",
            "server": ("test.execute-api.us-west-2.amazonaws.com", 80),
            "type": "http",
        }
        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, enable_lifespan=False)
    response = handler(mock_http_event, {})
    assert response == {
        "statusCode": 200,
        "isBase64Encoded": False,
        "headers": {"content-type": "text/plain; charset=utf-8"},
        "body": "Hello, world!",
    }
github erm / mangum / tests / test_websockets.py View on Github external
}

    handler = Mangum(app, enable_lifespan=False)
    with mock.patch(
        "mangum.protocols.websockets.ASGIWebSocketCycle.send_data"
    ) as send_data:
        send_data.return_value = None
        response = handler(mock_ws_send_event, {})
        assert response == {
            "body": "OK",
            "headers": {"content-type": "text/plain; charset=utf-8"},
            "isBase64Encoded": False,
            "statusCode": 200,
        }

    handler = Mangum(app, enable_lifespan=False)
    response = handler(mock_ws_disconnect_event, {})
    assert response == {
        "body": "OK",
        "headers": {"content-type": "text/plain; charset=utf-8"},
        "isBase64Encoded": False,
        "statusCode": 200,
    }
github erm / mangum / tests / test_websockets.py View on Github external
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
    )

    async def app(scope, receive, send):
        await send({"type": "websocket.send", "text": "Hello world!"})

    handler = Mangum(app, enable_lifespan=False)
    response = handler(mock_ws_connect_event, {})
    assert response == {
        "body": "OK",
        "headers": {"content-type": "text/plain; charset=utf-8"},
        "isBase64Encoded": False,
        "statusCode": 200,
    }

    handler = Mangum(app, enable_lifespan=False)

    with pytest.raises(RuntimeError):
        with mock.patch(
            "mangum.protocols.websockets.ASGIWebSocketCycle.send_data"
        ) as send_data:
            send_data.return_value = None
            handler(mock_ws_send_event, {})
github erm / mangum / tests / test_http.py View on Github external
def test_http_exception(mock_http_event) -> None:
    async def app(scope, receive, send):
        await send({"type": "http.response.start", "status": 200})
        raise Exception()
        await send({"type": "http.response.body", "body": b"1", "more_body": True})

    handler = Mangum(app, enable_lifespan=False)
    response = handler(mock_http_event, {})

    assert response == {
        "body": "Internal Server Error",
        "headers": {"content-type": "text/plain; charset=utf-8"},
        "isBase64Encoded": False,
        "statusCode": 500,
    }
github erm / mangum / tests / test_lifespan.py View on Github external
nonlocal startup_complete
        startup_complete = True

    @app.on_event("shutdown")
    async def on_shutdown():
        nonlocal shutdown_complete
        shutdown_complete = True

    @app.route(path)
    def homepage(request):
        return PlainTextResponse("Hello, world!")

    assert not startup_complete
    assert not shutdown_complete

    handler = Mangum(app)
    mock_http_event["body"] = None

    assert startup_complete
    assert not shutdown_complete

    response = handler(mock_http_event, {})

    assert response == {
        "statusCode": 200,
        "isBase64Encoded": False,
        "headers": {
            "content-length": "13",
            "content-type": "text/plain; charset=utf-8",
        },
        "body": "Hello, world!",
    }
github erm / asgi-examples / mangum / app / asgi.py View on Github external
# @app.route("/hello")
# def homepage(request):
#     return PlainTextResponse("hello world!")


# -- Quart -- #
# app = Quart(__name__)


# @app.route("/hello")
# async def hello():
#     return "hello world!"


handler = Mangum(app, debug=True)
github erm / guitarlette / src / app.py View on Github external
def transpose(content: str = Body(...), degree: int = Body(...)):
    parser = Song(content, degree=degree)
    return {"content": parser.text, "html_content": parser.html}


@app.post("/download")
def download(request: Request, content: str = Form(...)) -> StreamingResponse:
    filename = f"guitarlette-{datetime.now().strftime('%Y-%m-%d-%H:%M:%S')}.txt"
    return StreamingResponse(
        BytesIO(content.encode()),
        headers={"Content-Disposition": f"attachment; filename={filename}"},
        media_type="text/plain",
    )


handler = Mangum(app, lifespan="off")