adjust http trigger

Signed-off-by: laminar <[email protected]>

Author: tpiperatgod

setup.py

@@ -52,11 +52,13 @@
         "grpcio==1.54.2",
         "flask>=1.0,<3.0",
         "click>=7.0,<9.0",
-        # "watchdog>=1.0.0,<2.0.0",
+        "uvicorn>=0.22.0",
         "gunicorn>=19.2.0,<21.0; platform_system!='Windows'",
         "cloudevents>=1.2.0,<2.0.0",
         "dapr>=1.10.0",
         "aiohttp==3.8.4",
+        "dapr-ext-grpc>=1.10.0",
+        "dapr-ext-fastapi>=1.10.0"
     ],
     entry_points={
         "console_scripts": [

src/functions_framework/__init__.py

@@ -11,126 +11,3 @@
 # 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 io
-import json
-import logging
-import sys
-
-import flask
-
-from cloudevents.http import from_http
-
-from functions_framework.exceptions import (
-    EventConversionException,
-    FunctionsFrameworkException,
-)
-from functions_framework.openfunction import dapr_output_middleware
-from functions_framework.runner import Runner
-
-MAX_CONTENT_LENGTH = 10 * 1024 * 1024
-
-_FUNCTION_STATUS_HEADER_FIELD = "X-Google-Status"
-_CRASH = "crash"
-
-_CLOUDEVENT_MIME_TYPE = "application/cloudevents+json"
-
-
-class _LoggingHandler(io.TextIOWrapper):
-    """Logging replacement for stdout and stderr in GCF Python 3.7."""
-
-    def __init__(self, level, stderr=sys.stderr):
-        io.TextIOWrapper.__init__(self, io.StringIO(), encoding=stderr.encoding)
-        self.level = level
-        self.stderr = stderr
-
-    def write(self, out):
-        payload = dict(severity=self.level, message=out.rstrip("\n"))
-        return self.stderr.write(json.dumps(payload) + "\n")
-
-
-def setup_logging():
-    logging.getLogger().setLevel(logging.INFO)
-    info_handler = logging.StreamHandler(sys.stdout)
-    info_handler.setLevel(logging.NOTSET)
-    info_handler.addFilter(lambda record: record.levelno <= logging.INFO)
-    logging.getLogger().addHandler(info_handler)
-
-    warn_handler = logging.StreamHandler(sys.stderr)
-    warn_handler.setLevel(logging.WARNING)
-    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())
-
-    return view_func
-
-
-def _run_cloud_event(function, request):
-    data = request.get_data()
-    event = from_http(request.headers, data)
-    function(event)
-
-
-
-def read_request(response):
-    """
-    Force the framework to read the entire request before responding, to avoid
-    connection errors when returning prematurely.
-    """
-
-    flask.request.get_data()
-    return response
-
-
-def crash_handler(e):
-    """
-    Return crash header to allow logging 'crash' message in logs.
-    """
-    return str(e), 500, {_FUNCTION_STATUS_HEADER_FIELD: _CRASH}
-
-
-class LazyWSGIApp:
-    """
-    Wrap the WSGI app in a lazily initialized wrapper to prevent initialization
-    at import-time
-    """
-
-    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 = Runner(self.target, self.source, self.signature_type, self.func_context, self.debug)
-        return self.app(*args, **kwargs)
-
-
-app = LazyWSGIApp()
-
-
-class DummyErrorHandler:
-    def __init__(self):
-        pass
-
-    def __call__(self, *args, **kwargs):
-        return self
-
-
-errorhandler = DummyErrorHandler()

src/functions_framework/_cli.py

@@ -20,32 +20,14 @@
 @click.command()
 @click.option("--target", envvar="FUNCTION_TARGET", type=click.STRING, required=True)
 @click.option("--source", envvar="FUNCTION_SOURCE", type=click.Path(), default=None)
-# @click.option(
-#     "--signature-type",
-#     envvar="FUNCTION_SIGNATURE_TYPE",
-#     type=click.Choice(["http", "event", "cloudevent"]),
-#     default="http",
-# )
 @click.option("--host", envvar="HOST", type=click.STRING, default="0.0.0.0")
 @click.option("--port", envvar="PORT", type=click.INT, default=8080)
 @click.option("--debug", envvar="DEBUG", is_flag=True)
 @click.option("--dry-run", envvar="DRY_RUN", is_flag=True)
 def _cli(target, source, host, port, debug, dry_run):
-    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:
-    #     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)
+    # fetch the context
+    context = _function_registry.get_openfunction_context('')
+
     runner = Runner(context, target, source, host, port, debug, dry_run)
     runner.run()
 

src/functions_framework/_function_registry.py

@@ -45,7 +45,7 @@ def __function_signature_rule__(context: UserContext):
     pass
 
 
-FUNCTION_SIGNATURE_RULE = inspect.Signature(__function_signature_rule__())
+FUNCTION_SIGNATURE_RULE = inspect.signature(__function_signature_rule__)
 
 
 def get_user_function(source, source_module, target):
@@ -67,12 +67,12 @@ def get_user_function(source, source_module, target):
             )
         )
 
-    if FUNCTION_SIGNATURE_RULE != inspect.Signature(function):
+    if FUNCTION_SIGNATURE_RULE != inspect.signature(function):
         raise InvalidFunctionSignatureException(
             "The function defined in file {source} as {target} needs to be of "
             "function signature {signature}, but got {target_signature}".format(
                 source=source, target=target, signature=FUNCTION_SIGNATURE_RULE,
-                target_signature=inspect.Signature(function))
+                target_signature=inspect.signature(function))
         )
 
     return function

src/functions_framework/context/function_context.py

@@ -63,6 +63,8 @@ def from_json(json_dct):
         for trigger in _dapr_triggers:
             dapr_triggers.append(DaprTrigger.from_json(trigger))
 
+        http_trigger = HTTPRoute.from_json(http_trigger)
+
         return FunctionContext(name, version, dapr_triggers, http_trigger,
                                inputs, outputs, states, pre_hooks, post_hooks, tracing)
 
@@ -105,6 +107,29 @@ def from_json(json_dct):
         return Component(component_name, component_type, topic, metadata, operation)
 
 
+class HTTPRoute(object):
+    """HTTP route."""
+
+    def __init__(self, port="", hostname="", rules=None):
+        self.port = port
+        self.hostname = hostname
+        self.rules = rules
+
+    def __str__(self):
+        return "{port: %s, hostname: %s, rules: %s}" % (
+            self.port,
+            self.hostname,
+            self.rules
+        )
+
+    @staticmethod
+    def from_json(json_dct):
+        port = json_dct.get('port', '')
+        hostnames = json_dct.get('route', {}).get('hostnames', '')
+        rules = json_dct.get('route', {}).get('rules', [])
+        return HTTPRoute(port, hostnames, rules)
+
+
 class DaprTrigger(object):
 
     def __init__(self, name, component_type, topic):

src/functions_framework/context/runtime_context.py

@@ -11,13 +11,13 @@
 # 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.
-from functions_framework.context.function_context import FunctionContext, Component
+from functions_framework.context.function_context import Component, FunctionContext
 
 
-class RuntimeContext(object):
+class RuntimeContext:
     """Context for runtime."""
 
-    def __int__(self, context: FunctionContext = None, logger=None):
+    def __init__(self, context: FunctionContext = None, logger=None):
         self.context = context
         self.logger = logger
 
@@ -36,6 +36,13 @@ def get_dapr_triggers(self):
         else:
             return []
 
+    def get_http_trigger(self):
+        """Get http trigger."""
+        if self.context:
+            return self.context.http_trigger
+        else:
+            return None
+
     def get_outputs(self) -> [Component]:
         if self.context and self.context.outputs:
             return self.context.outputs

src/functions_framework/context/user_context.py

@@ -25,14 +25,15 @@
 class UserContext(object):
     """Context for user."""
 
-    def __int__(self, runtime_context: RuntimeContext = None,
-                binding_request=None, topic_event=None, logger=None):
+    def __init__(self, runtime_context: RuntimeContext = None,
+                binding_request=None, topic_event=None, http_request=None, logger=None):
         self.runtime_context = runtime_context
         self.logger = logger
         self.out = FunctionOut(0, None, "", {})
         self.dapr_client = None
         self.__binding_request = binding_request
         self.__topic_event = topic_event
+        self.__http_request = http_request
         self.__init_dapr_client()
 
     def __init_dapr_client(self):
@@ -49,6 +50,9 @@ def get_binding_request(self):
     def get_topic_event(self):
         return copy.deepcopy(self.__topic_event)
 
+    def get_http_request(self):
+        return self.__http_request
+
     @exception_handler
     def send(self, output_name, data):
         """Send data to specify output component.

src/functions_framework/log.py

@@ -46,9 +46,3 @@ def initialize_logger(name=None, level=logging.DEBUG):
 
 # initialize logger
 logger = initialize_logger(__name__, logging.INFO)
-
-# test logger
-logger.debug("This is a debug message")
-logger.info("This is an info message")
-logger.warning("This is a warning message")
-logger.error("This is an error message")

src/functions_framework/runner.py

@@ -21,6 +21,7 @@
 from functions_framework.context.runtime_context import RuntimeContext
 from functions_framework.exceptions import MissingSourceException
 from functions_framework.triggers.dapr_trigger.dapr import DaprTriggerHandler
+from functions_framework.triggers.http_trigger.http import HTTPTriggerHandler
 
 
 class Runner:
@@ -31,13 +32,13 @@ def __init__(self, context: FunctionContext, target=None, source=None,
         self.context = context
         self.user_function = None
         self.request = None
-        self.app = App()
         self.host = host
         self.port = port
         self.debug = debug
         self.dry_run = dry_run
         self.logger = None
         self.load_user_function()
+        self.init_logger()
 
     def load_user_function(self):
         _target = _function_registry.get_function_target(self.target)
@@ -63,11 +64,14 @@ def init_logger(self):
 
     def run(self):
         # convert to runtime context
-        runtime_context = RuntimeContext().__int__(self.context, self.logger)
+        runtime_context = RuntimeContext(self.context, self.logger)
+
+        _trigger = runtime_context.get_http_trigger()
+        if _trigger:
+            http_trigger = HTTPTriggerHandler(self.context.port, _trigger, self.source, self.target, self.user_function)
+            http_trigger.start(runtime_context, logger=self.logger)
 
         _triggers = runtime_context.get_dapr_triggers()
         if _triggers:
-            dapr_trigger = DaprTriggerHandler(self.context.port, self.app, _triggers, self.user_function)
-            dapr_trigger.start(runtime_context)
-
-        self.app.run(self.port)
+            dapr_trigger = DaprTriggerHandler(self.context.port, _triggers, self.user_function)
+            dapr_trigger.start(runtime_context, logger=self.logger)

src/functions_framework/triggers/dapr_trigger/dapr.py

@@ -25,10 +25,11 @@
 
 
 class DaprTriggerHandler(TriggerHandler):
-    def __init__(self, port, app: App = None, triggers: [DaprTrigger] = None, user_function=None):
+    """Handle dapr trigger."""
+    def __init__(self, port, triggers: [DaprTrigger] = None, user_function=None):
         self.port = port
         self.triggers = triggers
-        self.app = app
+        self.app = App()
         self.user_function = user_function
 
     def start(self, context: RuntimeContext, logger=None):
@@ -40,7 +41,7 @@ def start(self, context: RuntimeContext, logger=None):
                 @self.app.binding(trigger.name)
                 def binding_handler(request: BindingRequest):
                     rt_ctx = deepcopy(context)
-                    user_ctx = UserContext().__int__(runtime_context=rt_ctx, binding_request=request, logger=logger)
+                    user_ctx = UserContext(runtime_context=rt_ctx, binding_request=request, logger=logger)
                     logging.basicConfig(level=logging.DEBUG)
                     logging.info('Received Message : ' + request.text())
                     self.user_function(user_ctx)
@@ -49,7 +50,9 @@ def binding_handler(request: BindingRequest):
                 @self.app.subscribe(pubsub_name=trigger.name, topic=trigger.topic)
                 def topic_handler(event: v1.Event):
                     rt_ctx = deepcopy(context)
-                    user_ctx = UserContext().__int__(runtime_context=rt_ctx, topic_event=event, logger=logger)
+                    user_ctx = UserContext(runtime_context=rt_ctx, topic_event=event, logger=logger)
                     logging.basicConfig(level=logging.DEBUG)
                     logging.info('Received Message : ' + event.data.__str__())
                     self.user_function(user_ctx)
+
+        self.app.run(self.port)