vercel__workflow/workbench/python/app.py

"""ASGI adapter that puts a Python workflow app behind the routes the
TypeScript e2e suite drives.

The Python SDK owns the workflow and manifest handlers, their public route
constants, and the abstract HTTP request/response types. This file only adapts
those handlers to a bare ASGI app; a framework integration would do the same
routing itself.

It does not translate any protocol. `LocalWorld.create_queue_handler`
already reads exactly the headers `@workflow/world-local` sends and answers
`{"ok": true}`, and both sides read `WORKFLOW_LOCAL_DATA_DIR` with byte-compatible
file formats.

Only the *first* workflow trigger of a run arrives over HTTP. `LocalWorld.queue`
publishes to an embedded in-process queue service, so every step invocation and
workflow continuation after that stays inside this process.
"""

from __future__ import annotations

import importlib
from typing import Any, AsyncIterator, Callable

import httpx
from vercel.workflow import ENDPOINT_PATH, MANIFEST_PATH, HTTPRequest, HTTPResponse
from vercel.workflow._internal import world as w

# The fixture module is named to match `workbench/example/workflows/99_e2e.ts`,
# which a plain `import` statement cannot express — a module name may not start
# with a digit. importlib has no such restriction.
FIXTURES_MODULE = "workflows.99_e2e"
fixtures = importlib.import_module(FIXTURES_MODULE)
registry = fixtures.app
flow_handler = registry.http_handler
manifest_handler = registry.manifest_handler


class AsgiRequest(HTTPRequest):
    """The `HTTPRequest` the SDK's flow handler expects, over an ASGI scope."""

    def __init__(self, scope: dict[str, Any], receive: Callable) -> None:
        self._scope = scope
        self._headers = httpx.Headers(
            [(k.decode("latin-1"), v.decode("latin-1")) for k, v in scope["headers"]]
        )
        self._receive = receive

    @property
    def method(self) -> str:
        return self._scope["method"]

    @property
    def url(self) -> str:
        # The request target as it arrived, which is what the health branch
        # splits: it reports `endpoint` from the path, so an origin here would
        # make this app claim a path it does not serve.
        query = self._scope.get("query_string", b"").decode("latin-1")
        return self._scope["path"] + (f"?{query}" if query else "")

    @property
    def headers(self) -> httpx.Headers:
        return self._headers

    async def aiter_bytes(self, chunk_size: int | None = None) -> AsyncIterator[bytes]:
        while True:
            message = await self._receive()
            if message["type"] != "http.request":
                break
            body = message.get("body") or b""
            if body:
                yield body
            if not message.get("more_body"):
                break


async def _send_response(send: Callable, response: HTTPResponse) -> None:
    await send(
        {
            "type": "http.response.start",
            "status": response.status,
            "headers": [
                (key.encode(), value.encode()) for key, value in response.headers.items()
            ]
            + [(b"content-length", str(len(response.body)).encode())],
        }
    )
    await send({"type": "http.response.body", "body": response.body})


async def app(scope: dict[str, Any], receive: Callable, send: Callable) -> None:
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await w.get_world().aclose()
                await send({"type": "lifespan.shutdown.complete"})
                return

    if scope["type"] != "http":
        return

    path = scope["path"]
    method = scope["method"]
    query = scope.get("query_string", b"").decode()

    if path == MANIFEST_PATH and method in ("GET", "HEAD"):
        response = await manifest_handler(AsgiRequest(scope, receive))
        await _send_response(send, response)
        return

    # Two things share this route, and the handler tells them apart itself:
    # a queue delivery (POST) and the `?__health` probe that dev-server port
    # discovery sends to decide whether a port is serving a workflow app
    # (HEAD, reading only the status). Since vercel-py #292 the probe is the
    # SDK's, matching `withHealthCheck` in `@workflow/core` down to the CORS
    # headers, so this app only routes. The other, stream-based `healthCheck()`
    # protocol needs nothing here at all: it arrives as a queue delivery, and
    # `workflow_handler` answers it before it looks for a run.
    if path == ENDPOINT_PATH and (method == "POST" or "__health" in query):
        response = await flow_handler(AsgiRequest(scope, receive))
        await _send_response(send, response)
        return

    await _send_response(
        send,
        HTTPResponse.json({"error": f"No route for {method} {path}"}, status=404),
    )