Consider adding support for more web frameworks

[xSAVIKx]

While Flask is a decent choice as a default implementation, it may be beneficial to allow using different frameworks as the baseline for a Cloud Function.

I assume the following ones could be some great candidates:

• https://fastapi.tiangolo.com/
• https://www.tornadoweb.org/en/stable/
• https://sanic.readthedocs.io/en/latest/

We're also considering having framework-specific implementations of CloudEvents (e.g. FastAPI uses Pydantic by default and there's an open PR for Pydantic support in CloudEvents).
This can probably be handled as "extras" for the main library.

Is it smth the maintainers may consider?

[jama22]

I think that's an interesting idea, but it touches on something we've been wrestling with inside the functions-framework team.

The fact that people know that we use Flask underneath the hood is a bit of an anti-goal for the project. Flask is a convenient starting point to handle a lot of the workflows, and we don't wish to become a meta-abstraction layer on top of Flask. Taken to the extreme, we also don't want to be an abstraction layer for the other web frameworks you mentioned as well.

Some of the reasons for this is effort related: there are 7 languages in the functions-frameworld world and managing a cohesive set of functionality across all 7 is difficult enough. Creating the same functionality but with support for different web events adds to that combinatorial complexity.

The other big issue is..why? This is probably where I"d love to get some feedback from you @xSAVIKx . I'm not sure how how you're using functions-framework, so its not immediately clear to me why it would be valuable to support those frameworks. Are you a fan of the plugins/middleware that they provide? Or maybe its something about the performance of the frameworks themselves? Any information on your usecase and why your project would benefit would help us understand more about how to improve the project overall

[torbendury]

@jama22 Although this might be a little off-topic, I still have a question related to your chosen frameework: Why is it bad that people consider the functions-framework for Python as an abstraction layer for a serverless approach to Flask?

[jama22]

I think that's a fair question @torbendury . There's a fine line we're trying to walk with the functions framework project. We want to focus on building abstractions to support a broad range of functions with all kinds of input and output types. HTTP is one of the more useful ones, and so are CloudEvents and PubSub topics. In the case of Python, we use Flask as a means to achieve that goal.

When we get requests like "can you surface support for from Flask?"; we try to determine if it is a useful / necessary feature for HTTP functions for all languages, or if it is a Flask-specific feature that's only supporting the behaviors of the Flask framework

functions-framework project is still relatively young, so we are usually open to requests to surface specific capabilities in the context of a HTTP function. It's also why we're resistant to supporting more HTTP frameworks because it doesn't directly contribute to that goal

[torbendury]

Hey @jama22 and thank you for explaining this! Also could be a good general disclaimer you might want to put into the general Python functions-framework README :) I understand that Flask is more a go-to tool for you than you are just-another-abstraction for Flask and think that it was a good decision in general. Keep up the good work, I'm really enjoying the framework so far!

[mavwolverine]

@jama22 The frameworks listed by @xSAVIKx are all the newer async frameworks.
Since they all claim to be much more performant than flask, people might be expecting the ability to run an async framework.

That said, from cloud functions perspective, don't know how much of a performance difference it will be.

[mavwolverine]

fastapi/fastapi#812

Reverse question on fastapi side

[RazCrimson]

Here is a workaround that I am currently using (with Mangum):

import asyncio
import logging
from dataclasses import dataclass

from flask import Request
from flask import Response
from flask import make_response
from mangum.protocols import HTTPCycle
from mangum.types import ASGI

# Disable info logs from magnum (as GCP already logs requests automatically)
logger = logging.getLogger("mangum.http")
logger.setLevel(logging.WARNING)


@dataclass(slots=True)
class GcpMangum:
    """
    Wrapper to allow GCP Cloud Functions' HTTP (Flask) events to interact with ASGI frameworks
    Offloads internals to Mangum while acting as a wrapper for flask compatability
    """

    app: ASGI

    def __call__(self, request: Request) -> Response:
        try:
            return self.asgi(request)
        except BaseException as e:
            raise e

    def asgi(self, request: Request) -> Response:
        environ = request.environ
        scope = {
            "type": "http",
            "server": (environ["SERVER_NAME"], environ["SERVER_PORT"]),
            "client": environ["REMOTE_ADDR"],
            "method": request.method,
            "path": request.path,
            "scheme": request.scheme,
            "http_version": "1.1",
            "root_path": "",
            "query_string": request.query_string,
            "headers": [[k.lower().encode(), v.encode()] for k, v in request.headers],
        }
        request_body = request.data or b""

        try:
            asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        http_cycle = HTTPCycle(scope, request_body)
        http_response = http_cycle(self.app)
        converted_headers = [(name.decode(), val.decode()) for name, val in http_response["headers"]]
        return make_response(http_response["body"], http_response["status"], converted_headers)

I use it like this:

app = FastAPI(...)
api_handler = GcpMangum(app)

@functions_framework.http
def api(request):
    return api_handler(request)

[michaelg-baringa]

@RazCrimson can you share a bit more about the configuration for the cloud function please? Did you set api_handler as the entry point? I am getting an error saying

functions_framework.exceptions.InvalidTargetTypeException: The function defined in file /workspace/main.py as 'handler' needs to be of type function. Got: invalid type <class 'gcp_mangum.GcpMangum'>

[RazCrimson]

@michaelg-baringa
We need to apply the flask transformation applied by functions_framework to make the snippet work.

So if you want to use api_handler as the entrypoint, the code should look like:

app = FastAPI(...)
api_handler = functions_framework.http(GcpMangum(app))

[michaelg-baringa]

I tried that but got an error during the build phase: "AttributeError: 'GcpMangum' object has no attribute 'name'. Did you mean: 'ne'?"

File "/layers/google.python.pip/pip/lib/python3.11/site-packages/functions_framework/init.py", line 115, in http_function_registry.REGISTRY_MAP[func.name] = (

So I went back to your original way and that worked for the build phase but was getting errors when calling the endpoint, something about content-length not being a string value.

2024-06-13 20:19:49.663
    raise TypeError('%r is not a string' % name)
2024-06-13 20:19:50.783
TypeError: b'content-length' is not a string

I've since tried another library, agraffe, which is working so likely going to go with that instead. Thanks for the help though!

[JeroenPeterBos]

@RazCrimson

Thanks for the workaround with Mangum.
I noticed the lifecycle management was not included so I have added that.

The following works for me:

"""
GCP by default uses Flask for HTTP requests.
Mangum is an ASGI framework that can be used to handle HTTP requests for AWS Lambda functions.
This file is a wrapper to allow GCP Cloud Functions' HTTP (Flask) events to interact with ASGI frameworks using Mangum.
"""

import asyncio
import logging
from dataclasses import dataclass
from contextlib import ExitStack


from flask import Request
from flask import Response
from flask import make_response
from mangum.protocols import HTTPCycle, LifespanCycle
from mangum.types import ASGI

# Disable info logs from magnum (as GCP already logs requests automatically)
log = logging.getLogger("mangum.http")
log.setLevel(logging.WARNING)


@dataclass(slots=True)
class MangumGCP:
    """
    Wrapper to allow GCP Cloud Functions' HTTP (Flask) events to interact with ASGI frameworks
    Offloads internals to Mangum while acting as a wrapper for flask compatability
    """

    app: ASGI
    lifespan: str = "auto"
    text_mime_types: list[str] | None = None
    exclude_headers: list[str] | None = None

    def __call__(self, request: Request) -> Response:
        try:
            return self.asgi(request)
        except BaseException as e:
            raise e

    def asgi(self, request: Request) -> Response:
        environ = request.environ
        scope = {
            "type": "http",
            "server": (environ["SERVER_NAME"], environ["SERVER_PORT"]),
            "client": environ["REMOTE_ADDR"],
            "method": request.method,
            "path": request.path,
            "scheme": request.scheme,
            "http_version": "1.1",
            "root_path": "",
            "query_string": request.query_string,
            "headers": [[k.lower().encode(), v.encode()] for k, v in request.headers],
        }
        request_body = request.data or b""

        try:
            asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        with ExitStack() as stack:
            if self.lifespan in ("auto", "on"):
                lifespan_cycle = LifespanCycle(self.app, self.lifespan)  # type: ignore
                stack.enter_context(lifespan_cycle)
                scope.update({"state": lifespan_cycle.lifespan_state.copy()})

            http_cycle = HTTPCycle(scope, request_body)
            http_response = http_cycle(self.app)

        # Optionally filter/exclude headers
        exclude_headers = set((self.exclude_headers or []))
        converted_headers = [
            (name.decode(), val.decode())
            for name, val in http_response["headers"]
            if name.decode().lower() not in exclude_headers
        ]
        return make_response(http_response["body"], http_response["status"], converted_headers)

@michaelg-baringa

To deploy this you can use the following in your main.py file and then use --entry-point=handler as an argument for your gcloud functions deploy command.

gcp_function_handler = MangumGCP(app)


@functions_framework.http
def handler(request):
    return gcp_function_handler(request)