feat: Introduce functions_framework.aio submodule that support async function execution.

Author: taeold

setup.py

@@ -57,6 +57,9 @@
         "cloudevents>=1.2.0,<2.0.0",
         "Werkzeug>=0.14,<4.0.0",
     ],
+    extras_require={
+        "async": ["starlette>=0.37.0,<1.0.0"],
+    },
     entry_points={
         "console_scripts": [
             "ff=functions_framework._cli:_cli",

src/functions_framework/aio/__init__.py

@@ -0,0 +1,244 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+import functools
+import inspect
+import os
+
+from typing import Any, Awaitable, Callable, Union
+
+from cloudevents.http import from_http
+from cloudevents.http.event import CloudEvent
+
+from functions_framework import _function_registry
+from functions_framework.exceptions import (
+    FunctionsFrameworkException,
+    MissingSourceException,
+)
+
+try:
+    from starlette.applications import Starlette
+    from starlette.exceptions import HTTPException
+    from starlette.requests import Request
+    from starlette.responses import JSONResponse, Response
+    from starlette.routing import Route
+except ImportError:
+    raise FunctionsFrameworkException(
+        "Starlette is not installed. Install the framework with the 'async' extra: "
+        "pip install functions-framework[async]"
+    )
+
+HTTPResponse = Union[
+    Response,  # Functions can return a full Starlette Response object
+    str,  # Str returns are wrapped in Response(result)
+    dict[Any, Any],  # Dict returns are wrapped in JSONResponse(result)
+    tuple[Any, int],  # Flask-style (content, status_code) supported
+    None,  # None raises HTTPException
+]
+
+_FUNCTION_STATUS_HEADER_FIELD = "X-Google-Status"
+_CRASH = "crash"
+
+CloudEventFunction = Callable[[CloudEvent], Union[None, Awaitable[None]]]
+HTTPFunction = Callable[[Request], Union[HTTPResponse, Awaitable[HTTPResponse]]]
+
+
+def cloud_event(func: CloudEventFunction) -> CloudEventFunction:
+    """Decorator that registers cloudevent as user function signature type."""
+    _function_registry.REGISTRY_MAP[func.__name__] = (
+        _function_registry.CLOUDEVENT_SIGNATURE_TYPE
+    )
+    if inspect.iscoroutinefunction(func):
+
+        @functools.wraps(func)
+        async def async_wrapper(*args, **kwargs):
+            return await func(*args, **kwargs)
+
+        return async_wrapper
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+def http(func: HTTPFunction) -> HTTPFunction:
+    """Decorator that registers http as user function signature type."""
+    _function_registry.REGISTRY_MAP[func.__name__] = (
+        _function_registry.HTTP_SIGNATURE_TYPE
+    )
+
+    if inspect.iscoroutinefunction(func):
+
+        @functools.wraps(func)
+        async def async_wrapper(*args, **kwargs):
+            return await func(*args, **kwargs)
+
+        return async_wrapper
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+async def _crash_handler(request, exc):
+    headers = {_FUNCTION_STATUS_HEADER_FIELD: _CRASH}
+    return Response(f"Internal Server Error: {exc}", status_code=500, headers=headers)
+
+
+def _http_func_wrapper(function, is_async):
+    @functools.wraps(function)
+    async def handler(request):
+        if is_async:
+            result = await function(request)
+        else:
+            result = await asyncio.to_thread(function, request)
+        if isinstance(result, str):
+            return Response(result)
+        elif isinstance(result, dict):
+            return JSONResponse(result)
+        elif isinstance(result, tuple) and len(result) == 2:
+            # Support Flask-style tuple response
+            content, status_code = result
+            return Response(content, status_code=status_code)
+        elif result is None:
+            raise HTTPException(status_code=500, detail="No response returned")
+        else:
+            return result
+
+    return handler
+
+
+def _cloudevent_func_wrapper(function, is_async):
+    @functools.wraps(function)
+    async def handler(request):
+        data = await request.body()
+
+        try:
+            event = from_http(request.headers, data)
+        except Exception as e:
+            raise HTTPException(
+                400, detail=f"Bad Request: Got CloudEvent exception: {repr(e)}"
+            )
+        if is_async:
+            await function(event)
+        else:
+            await asyncio.to_thread(function, event)
+        return Response("OK")
+
+    return handler
+
+
+async def _handle_not_found(request: Request):
+    raise HTTPException(status_code=404, detail="Not Found")
+
+
+def create_asgi_app(target=None, source=None, signature_type=None):
+    """Create an ASGI application for the function.
+
+    Args:
+        target: The name of the target function to invoke
+        source: The source file containing the function
+        signature_type: The signature type of the function
+                       ('http', 'event', 'cloudevent', or 'typed')
+
+    Returns:
+        A Starlette ASGI application instance
+    """
+    target = _function_registry.get_function_target(target)
+    source = _function_registry.get_function_source(source)
+
+    if not os.path.exists(source):
+        raise MissingSourceException(
+            f"File {source} that is expected to define function doesn't exist"
+        )
+
+    source_module, spec = _function_registry.load_function_module(source)
+    spec.loader.exec_module(source_module)
+    function = _function_registry.get_user_function(source, source_module, target)
+    signature_type = _function_registry.get_func_signature_type(target, signature_type)
+
+    is_async = inspect.iscoroutinefunction(function)
+    routes = []
+    if signature_type == _function_registry.HTTP_SIGNATURE_TYPE:
+        http_handler = _http_func_wrapper(function, is_async)
+        routes.append(
+            Route(
+                "/",
+                endpoint=http_handler,
+                methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"],
+            ),
+        )
+        routes.append(Route("/robots.txt", endpoint=_handle_not_found, methods=["GET"]))
+        routes.append(
+            Route("/favicon.ico", endpoint=_handle_not_found, methods=["GET"])
+        )
+        routes.append(
+            Route(
+                "/{path:path}",
+                http_handler,
+                methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"],
+            )
+        )
+    elif signature_type == _function_registry.CLOUDEVENT_SIGNATURE_TYPE:
+        cloudevent_handler = _cloudevent_func_wrapper(function, is_async)
+        routes.append(Route("/{path:path}", cloudevent_handler, methods=["POST"]))
+        routes.append(Route("/", cloudevent_handler, methods=["POST"]))
+    elif signature_type == _function_registry.TYPED_SIGNATURE_TYPE:
+        raise FunctionsFrameworkException(
+            f"ASGI server does not support typed events (signature type: '{signature_type}'). "
+        )
+    elif signature_type == _function_registry.BACKGROUNDEVENT_SIGNATURE_TYPE:
+        raise FunctionsFrameworkException(
+            f"ASGI server does not support legacy background events (signature type: '{signature_type}'). "
+            "Use 'cloudevent' signature type instead."
+        )
+    else:
+        raise FunctionsFrameworkException(
+            f"Unsupported signature type for ASGI server: {signature_type}"
+        )
+
+    exception_handlers = {
+        500: _crash_handler,
+    }
+    app = Starlette(routes=routes, exception_handlers=exception_handlers)
+    return app
+
+
+class LazyASGIApp:
+    """
+    Wrap the ASGI app in a lazily initialized wrapper to prevent initialization
+    at import-time
+    """
+
+    def __init__(self, target=None, source=None, signature_type=None):
+        self.target = target
+        self.source = source
+        self.signature_type = signature_type
+
+        self.app = None
+        self._app_initialized = False
+
+    async def __call__(self, scope, receive, send):
+        if not self._app_initialized:
+            self.app = create_asgi_app(self.target, self.source, self.signature_type)
+            self._app_initialized = True
+        await self.app(scope, receive, send)
+
+
+app = LazyASGIApp()