feat: init binding and async (#1)

* test: fix unit tests (GoogleCloudPlatform#181)

Fix failing assert to look for a different error message, likely due to
a change in CloudEvents error message.

Also add tests so that code coverage is back at 100%.

Signed-off-by: Kehui Li <[email protected]>

Author: kehuili

README.md

@@ -5,15 +5,17 @@
 [![Python unit CI][ff_python_unit_img]][ff_python_unit_link] [![Python lint CI][ff_python_lint_img]][ff_python_lint_link] [![Python conformace CI][ff_python_conformance_img]][ff_python_conformance_link]
 
 An open source FaaS (Function as a service) framework for writing portable
-Python functions -- brought to you by the Google Cloud Functions team.
+Python functions.
 
 The Functions Framework lets you write lightweight functions that run in many
 different environments, including:
 
+*   [OpenFunction](https://github.com/OpenFunction/OpenFunction)
+*   [Knative](https://github.com/knative/)-based environments
+*   [Dapr](https://dapr.io/)-based environments
 *   [Google Cloud Functions](https://cloud.google.com/functions/)
 *   Your local development machine
 *   [Cloud Run and Cloud Run for Anthos](https://cloud.google.com/run/)
-*   [Knative](https://github.com/knative/)-based environments
 
 The framework allows you to go from:
 
@@ -292,6 +294,22 @@ https://cloud.google.com/functions/docs/tutorials/pubsub#functions_helloworld_pu
 
 ## Run your function on serverless platforms
 
+### Container environments based on Knative
+
+The Functions Framework is designed to be compatible with Knative environments. Build and deploy your container to a Knative environment.
+
+### OpenFunction
+
+![OpenFunction Platform Overview](https://openfunction.dev/openfunction-0.5-architecture.png)
+
+Besides Knative function support, one notable feature of OpenFunction is embracing Dapr system, so far Dapr pub/sub and bindings have been support.
+
+Dapr bindings allows you to trigger your applications or services with events coming in from external systems, or interface with external systems. OpenFunction [0.6.0 release](https://openfunction.dev/blog/2022/03/25/announcing-openfunction-0.6.0-faas-observability-http-trigger-and-more/) adds Dapr output bindings to its synchronous functions which enables HTTP triggers for asynchronous functions. For example, synchronous functions backed by the Knative runtime can now interact with middlewares defined by Dapr output binding or pub/sub, and an asynchronous function will be triggered by the events sent from the synchronous function.
+
+Asynchronous function introduces Dapr pub/sub to provide a platform-agnostic API to send and receive messages. A typical use case is that you can leverage synchronous functions to receive an event in plain JSON or Cloud Events format, and then send the received event to a Dapr output binding or pub/sub component, most likely a message queue (e.g. Kafka, NATS Streaming, GCP PubSub, MQTT). Finally, the asynchronous function could be triggered from the message queue.
+
+More details would be brought up to you in some quickstart samples, stay tuned.
+
 ### Google Cloud Functions
 
 This Functions Framework is based on the [Python Runtime on Google Cloud Functions](https://cloud.google.com/functions/docs/concepts/python-runtime).

docs/async-server.puml

@@ -0,0 +1,59 @@
+@startuml Async Server
+
+box Function Process in Local Environment or Container
+control ENTRYPOINT
+participant "~__main__" as Main
+participant AsyncServer
+participant DaprServer
+participant gRPCServer [
+  Web Server
+  ----
+  ""gprc.server""
+]
+end box
+
+entity "Dapr Sidecar " as DaprSidecar
+
+== OpenFunction Serving ==
+
+ENTRYPOINT -> Main ** : execute
+note over ENTRYPOINT, Main: Pass through __CLI arguments__ and \ncontainer __environment variables__
+
+Main -> Main : load user function file
+note left: ""function (ctx, data) {}""
+
+Main -> AsyncServer ** : create
+note over Main, AsyncServer: Hand over __user function__ and __context__
+
+AsyncServer -> DaprServer ** : ""new""
+note over AsyncServer, DaprServer: Extract __port__ from __context__ and pass down
+
+DaprServer -> gRPCServer ** : ""new""
+|||
+DaprServer --> DaprSidecar : Waiting till Dapr sidecar started
+...
+AsyncServer -> DaprServer : register __user function__ as handler \nfor each of __inputs__ in __context__
+DaprServer -> gRPCServer : add routes for Dapr style \nsubscriptions and input bindings
+
+...
+
+== OpenFunction Triggering ==
+
+DaprSidecar <-- : sub / input data
+
+DaprSidecar -> gRPCServer ++ : Dapr request with "data"
+
+gRPCServer -> gRPCServer ++ : invoke user function
+
+alt 
+  gRPCServer -> DaprSidecar ++ : publish data or invoke output binding
+  DaprSidecar --> gRPCServer -- : execution result
+end
+
+return
+
+return server app response
+
+...
+
+@enduml

docs/http-binding.puml

@@ -0,0 +1,73 @@
+@startuml HTTP Binding
+
+box Function Process in Local Environment or Container
+control ENTRYPOINT
+participant "~__main__" as Main
+participant HTTPServer
+participant Server [
+  Web Server
+  ----
+  ""Flask/Gunicorn""
+]
+participant Middleware
+participant "User Function" as UserFunction
+participant DaprClient
+end box
+
+entity "Dapr Sidecar " as DaprSidecar
+
+== OpenFunction Serving ==
+
+ENTRYPOINT -> Main ** : execute
+note over ENTRYPOINT, Main: Pass through __CLI arguments__ and \ncontainer __environment variables__
+
+Main -> Main : load user fnction file
+note left: ""function (request) {}"" 
+
+Main -> HTTPServer ** : create
+note over Main, HTTPServer: Hand over __user function__, __function type__ \nand __context__ parsed from env variables 
+
+HTTPServer -> Server ** : new
+note over Server: Depend on debug mode
+
+HTTPServer -> Middleware ** : new
+HTTPServer -> Server : use Middleware
+note over HTTPServer, Server: Pass context to middleware
+|||
+HTTPServer -> Server : use others middlewares
+|||
+HTTPServer -> UserFunction ** : wrap user function
+note over HTTPServer, UserFunction: Register as HTTP or CloudEvent Function 
+HTTPServer -> Server : bind wrapper to "/*" route
+
+...
+
+== OpenFunction Invocation ==
+
+[-> Server ++ : HTTP request to "/"
+
+Server -> UserFunction ++ : execute user function
+UserFunction --> Server -- : return execution result "data"
+
+alt ""runtime"" = ""knative"" and ""outputs"" is not empty
+  Server -> Middleware ++ : invoke Middleware
+
+  Middleware -> DaprClient ** : new
+
+  loop each OpenFunction Output 
+    Middleware -> DaprClient ++ : send "data"
+
+    DaprClient -> DaprSidecar ++ : invoke binding or publication with "data"
+    DaprSidecar --> DaprClient -- : return result 
+
+    DaprClient --> Middleware -- : forward result
+  end
+
+  Middleware --> Server -- : return "data" as response
+end
+
+[<- Server -- : send response
+
+...
+
+@enduml

setup.py

@@ -25,13 +25,13 @@
 
 setup(
     name="functions-framework",
-    version="3.0.0",
-    description="An open source FaaS (Function as a service) framework for writing portable Python functions -- brought to you by the Google Cloud Functions team.",
+    version="3.1.0",
+    description="An open source FaaS (Function as a service) framework for writing portable Python functions.",
     long_description=long_description,
     long_description_content_type="text/markdown",
-    url="https://github.com/googlecloudplatform/functions-framework-python",
-    author="Google LLC",
-    author_email="[email protected]",
+    url="https://github.com/OpenFunction/functions-framework-python",
+    author="OpenFunction",
+    author_email="[email protected]",
     classifiers=[
         "Development Status :: 3 - Alpha",
         "Intended Audience :: Developers",
@@ -55,6 +55,7 @@
         "watchdog>=1.0.0,<2.0.0",
         "gunicorn>=19.2.0,<21.0; platform_system!='Windows'",
         "cloudevents>=1.2.0,<2.0.0",
+        "dapr>=1.6.0",
     ],
     entry_points={
         "console_scripts": [

src/functions_framework/__init__.py

@@ -34,6 +34,8 @@
     MissingSourceException,
 )
 from google.cloud.functions.context import Context
+from openfunction.dapr_output_middleware import dapr_output_middleware
+from openfunction.async_server import AsyncApp
 
 MAX_CONTENT_LENGTH = 10 * 1024 * 1024
 
@@ -94,6 +96,11 @@ def setup_logging():
     logging.getLogger().addHandler(warn_handler)
 
 
+def setup_logging_level(debug):
+    if debug:
+        logging.getLogger().setLevel(logging.DEBUG)
+
+
 def _http_view_func_wrapper(function, request):
     def view_func(path):
         return function(request._get_current_object())
@@ -175,7 +182,7 @@ def view_func(path):
     return view_func
 
 
-def _configure_app(app, function, signature_type):
+def _configure_app(app, function, signature_type, func_context):
     # Mount the function at the root. Support GCF's default path behavior
     # Modify the url_map and view_functions directly here instead of using
     # add_url_rule in order to create endpoints that route all methods
@@ -189,6 +196,7 @@ def _configure_app(app, function, signature_type):
         app.view_functions["run"] = _http_view_func_wrapper(function, flask.request)
         app.view_functions["error"] = lambda: flask.abort(404, description="Not Found")
         app.after_request(read_request)
+        app.after_request(dapr_output_middleware(func_context))
     elif signature_type == _function_registry.BACKGROUNDEVENT_SIGNATURE_TYPE:
         app.url_map.add(
             werkzeug.routing.Rule(
@@ -241,8 +249,31 @@ def crash_handler(e):
     """
     return str(e), 500, {_FUNCTION_STATUS_HEADER_FIELD: _CRASH}
 
+def create_async_app(target=None, source=None, func_context=None, debug=False):
+    target = _function_registry.get_function_target(target)
+    source = _function_registry.get_function_source(source)
+
+    if not os.path.exists(source):
+        raise MissingSourceException(
+            "File {source} that is expected to define function doesn't exist".format(
+                source=source
+            )
+        )
+
+    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)
+
+    setup_logging_level(debug)
+
+    async_app = AsyncApp(func_context)
+    async_app.bind(function)
+
+    return async_app.app
+
 
-def create_app(target=None, source=None, signature_type=None):
+def create_app(target=None, source=None, signature_type=None, func_context=None, debug=False):
     target = _function_registry.get_function_target(target)
     source = _function_registry.get_function_source(source)
 
@@ -282,6 +313,8 @@ def handle_none(rv):
         sys.stdout = _LoggingHandler("INFO", sys.stderr)
         sys.stderr = _LoggingHandler("ERROR", sys.stderr)
         setup_logging()
+        
+    setup_logging_level(debug)
 
     # Execute the module, within the application context
     with _app.app_context():
@@ -291,7 +324,7 @@ def handle_none(rv):
     signature_type = _function_registry.get_func_signature_type(target, signature_type)
     function = _function_registry.get_user_function(source, source_module, target)
 
-    _configure_app(_app, function, signature_type)
+    _configure_app(_app, function, signature_type, func_context)
 
     return _app
 
@@ -302,21 +335,23 @@ class LazyWSGIApp:
     at import-time
     """
 
-    def __init__(self, target=None, source=None, signature_type=None):
+    def __init__(self, target=None, source=None, signature_type=None, func_context=None, debug=False):
         # Support HTTP frameworks which support WSGI callables.
         # Note: this ability is currently broken in Gunicorn 20.0, and
         # environment variables should be used for configuration instead:
         # https://github.com/benoitc/gunicorn/issues/2159
         self.target = target
         self.source = source
         self.signature_type = signature_type
+        self.func_context = func_context
+        self.debug = debug
 
         # Placeholder for the app which will be initialized on first call
         self.app = None
 
     def __call__(self, *args, **kwargs):
         if not self.app:
-            self.app = create_app(self.target, self.source, self.signature_type)
+            self.app = create_app(self.target, self.source, self.signature_type, self.func_context, self.debug)
         return self.app(*args, **kwargs)
 
 

src/functions_framework/_cli.py

@@ -16,9 +16,9 @@
 
 import click
 
-from functions_framework import create_app
+from functions_framework import create_app, create_async_app
 from functions_framework._http import create_server
-
+from functions_framework import _function_registry
 
 @click.command()
 @click.option("--target", envvar="FUNCTION_TARGET", type=click.STRING, required=True)
@@ -34,10 +34,24 @@
 @click.option("--debug", envvar="DEBUG", is_flag=True)
 @click.option("--dry-run", envvar="DRY_RUN", is_flag=True)
 def _cli(target, source, signature_type, host, port, debug, dry_run):
-    app = create_app(target, source, signature_type)
-    if dry_run:
-        click.echo("Function: {}".format(target))
-        click.echo("URL: http://{}:{}/".format(host, port))
-        click.echo("Dry run successful, shutting down.")
+    context = _function_registry.get_openfunction_context(None)
+    
+    # determine if async or knative
+    if context and context.is_runtime_async():
+        app = create_async_app(target, source, context, debug)
+        if dry_run:
+            run_dry(target, host, port)
+        else:
+            app.run(context.port)
     else:
-        create_server(app, debug).run(host, port)
+        app = create_app(target, source, signature_type, context, debug)
+        if dry_run:
+            run_dry(target, host, port)
+        else:
+            create_server(app, debug).run(host, port)
+
+
+def run_dry(target, host, port):
+    click.echo("Function: {}".format(target))
+    click.echo("URL: http://{}:{}/".format(host, port))
+    click.echo("Dry run successful, shutting down.")