feat: Support strongly typed functions signature

src/functions_framework/__init__.py

@@ -13,20 +13,25 @@
 # limitations under the License.
 
 import functools
+import inspect
 import io
 import json
 import logging
 import os.path
 import pathlib
 import sys
+import types
+
+from inspect import signature
+from typing import Type
 
 import cloudevents.exceptions as cloud_exceptions
 import flask
 import werkzeug
 
 from cloudevents.http import from_http, is_binary
 
-from functions_framework import _function_registry, event_conversion
+from functions_framework import _function_registry, _typed_event, event_conversion
 from functions_framework.background_event import BackgroundEvent
 from functions_framework.exceptions import (
  EventConversionException,
@@ -67,6 +72,33 @@ def wrapper(*args, **kwargs):
  return wrapper
 
 
+def typed(*args):
+ def _typed(func):
+ _typed_event.register_typed_event(input_type, func)
+
+ @functools.wraps(func)
+ def wrapper(*args, **kwargs):
+ return func(*args, **kwargs)
+
+ return wrapper
+
+ # no input type provided as a parameter, we need to use reflection
+ # e.g function declaration:
+ # @typed
+ # def myfunc(x:input_type)
+ if len(args) == 1 and isinstance(args[0], types.FunctionType):
+ input_type = None
+ return _typed(args[0])
+
+ # input type provided as a parameter to the decorator
+ # e.g. function declaration
+ # @typed(input_type)
+ # def myfunc(x)
+ else:
+ input_type = args[0]
+ return _typed
+
+
 def http(func):
  """Decorator that registers http as user function signature type."""
  _function_registry.REGISTRY_MAP[
@@ -106,6 +138,26 @@ def _run_cloud_event(function, request):
  function(event)
 
 
+def _typed_event_func_wrapper(function, request, inputType: Type):
+ def view_func(path):
+ try:
+ data = request.get_json()
+ input = inputType.from_dict(data)
+ response = function(input)
+ if response is None:
+ return "", 200
+ if response.__class__.__module__ == "builtins":
+ return response
+ _typed_event._validate_return_type(response)
+ return json.dumps(response.to_dict())
+ except Exception as e:
+ raise FunctionsFrameworkException(
+ "Function execution failed with the error"
+ ) from e
+
+ return view_func
+
+
 def _cloud_event_view_func_wrapper(function, request):
  def view_func(path):
  ce_exception = None
@@ -216,6 +268,21 @@ def _configure_app(app, function, signature_type):
  app.view_functions[signature_type] = _cloud_event_view_func_wrapper(
  function, flask.request
  )
+ elif signature_type == _function_registry.TYPED_SIGNATURE_TYPE:
+ app.url_map.add(
+ werkzeug.routing.Rule(
+ "/", defaults={"path": ""}, endpoint=signature_type, methods=["POST"]
+ )
+ )
+ app.url_map.add(
+ werkzeug.routing.Rule(
+ "/<path:path>", endpoint=signature_type, methods=["POST"]
+ )
+ )
+ input_type = _function_registry.get_func_input_type(function.__name__)
+ app.view_functions[signature_type] = _typed_event_func_wrapper(
+ function, flask.request, input_type
+ )
  else:
  raise FunctionsFrameworkException(
  "Invalid signature type: {signature_type}".format(

src/functions_framework/_cli.py

@@ -26,7 +26,7 @@
 @click.option(
  "--signature-type",
  envvar="FUNCTION_SIGNATURE_TYPE",
- type=click.Choice(["http", "event", "cloudevent"]),
+ type=click.Choice(["http", "event", "cloudevent", "typed"]),
  default="http",
 )
 @click.option("--host", envvar="HOST", type=click.STRING, default="0.0.0.0")

src/functions_framework/_function_registry.py

@@ -16,6 +16,9 @@
 import sys
 import types
 
+from re import T
+from typing import Type
+
 from functions_framework.exceptions import (
  InvalidConfigurationException,
  InvalidTargetTypeException,
@@ -28,11 +31,16 @@
 HTTP_SIGNATURE_TYPE = "http"
 CLOUDEVENT_SIGNATURE_TYPE = "cloudevent"
 BACKGROUNDEVENT_SIGNATURE_TYPE = "event"
+TYPED_SIGNATURE_TYPE = "typed"
 
 # REGISTRY_MAP stores the registered functions.
 # Keys are user function names, values are user function signature types.
 REGISTRY_MAP = {}
 
+# INPUT_TYPE_MAP stores the input type of the typed functions.
+# Keys are the user function name, values are the type of the function input
+INPUT_TYPE_MAP = {}
+
 
 def get_user_function(source, source_module, target):
  """Returns user function, raises exception for invalid function."""
@@ -120,3 +128,8 @@ def get_func_signature_type(func_name: str, signature_type: str) -> str:
  if os.environ.get("ENTRY_POINT"):
  os.environ["FUNCTION_TRIGGER_TYPE"] = sig_type
  return sig_type
+
+
+def get_func_input_type(func_name: str) -> Type:
+ registered_type = INPUT_TYPE_MAP[func_name] if func_name in INPUT_TYPE_MAP else ""
+ return registered_type

src/functions_framework/_typed_event.py

@@ -0,0 +1,105 @@
+# Copyright 2022 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 inspect
+
+from inspect import signature
+
+from functions_framework import _function_registry
+from functions_framework.exceptions import FunctionsFrameworkException
+
+"""Registers user function in the REGISTRY_MAP and the INPUT_TYPE_MAP.
+Also performs some validity checks for the input type of the function
+
+Args:
+ decorator_type: The type provided by the @typed(input_type) decorator
+ func: User function
+"""
+
+
+def register_typed_event(decorator_type, func):
+ try:
+ sig = signature(func)
+ annotation_type = list(sig.parameters.values())[0].annotation
+ input_type = _select_input_type(decorator_type, annotation_type)
+ _validate_input_type(input_type)
+ except IndexError:
+ raise FunctionsFrameworkException(
+ "Function signature is missing an input parameter."
+ "The function should be defined as 'def your_fn(in: inputType)'"
+ )
+ except Exception as e:
+ raise FunctionsFrameworkException(
+ "Functions using the @typed decorator must provide "
+ "the type of the input parameter by specifying @typed(inputType) and/or using python "
+ "type annotations 'def your_fn(in: inputType)'"
+ )
+
+ _function_registry.INPUT_TYPE_MAP[func.__name__] = input_type
+ _function_registry.REGISTRY_MAP[
+ func.__name__
+ ] = _function_registry.TYPED_SIGNATURE_TYPE
+
+
+""" Checks whether the response type of the typed function has a to_dict method"""
+
+
+def _validate_return_type(response):
+ if not (hasattr(response, "to_dict") and callable(getattr(response, "to_dict"))):
+ raise AttributeError(
+ "The type {response} does not have the required method called "
+ " 'to_dict'.".format(response=type(response))
+ )
+
+
+"""Selects the input type for the typed function provided through the @typed(input_type)
+decorator or through the parameter annotation in the user function
+"""
+
+
+def _select_input_type(decorator_type, annotation_type):
+ if decorator_type == None and annotation_type is inspect._empty:
+ raise TypeError(
+ "The function defined does not contain Type of the input object."
+ )
+
+ if (
+ decorator_type != None
+ and annotation_type is not inspect._empty
+ and decorator_type != annotation_type
+ ):
+ raise TypeError(
+ "The object type provided via 'typed' decorator: '{decorator_type}'"
+ "is different than the one specified by the function parameter's type annotation : '{annotation_type}'.".format(
+ decorator_type=decorator_type, annotation_type=annotation_type
+ )
+ )
+
+ if decorator_type == None:
+ return annotation_type
+ return decorator_type
+
+
+"""Checks for the from_dict method implementation in the input type class"""
+
+
+def _validate_input_type(input_type):
+ if not (
+ hasattr(input_type, "from_dict") and callable(getattr(input_type, "from_dict"))
+ ):
+ raise AttributeError(
+ "The type {decorator_type} does not have the required method called "
+ " 'from_dict'.".format(decorator_type=input_type)
+ )

tests/test_functions/typed_events/mismatch_types.py

@@ -0,0 +1,43 @@
+# Copyright 2022 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.
+
+"""Function used to test handling functions using typed decorators."""
+
+import flask
+
+import functions_framework
+
+
+class TestType1:
+ name: str
+ age: int
+
+ def __init__(self, name: str, age: int) -> None:
+ self.name = name
+ self.age = age
+
+
+class TestType2:
+ name: str
+
+ def __init__(self, name: str) -> None:
+ self.name = name
+
+
+@functions_framework.typed(TestType2)
+def function_typed_mismatch_types(test_type: TestType1):
+ valid_event = test_type.name == "john" and test_type.age == 10
+ if not valid_event:
+ raise Exception("Received invalid input")
+ return test_type

tests/test_functions/typed_events/missing_from_dict.py

@@ -0,0 +1,55 @@
+# Copyright 2022 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.
+
+"""Function used to test handling functions using typed decorators."""
+from typing import Any, TypeVar
+
+import flask
+
+import functions_framework
+
+T = TypeVar("T")
+
+
+def from_str(x: Any) -> str:
+ assert isinstance(x, str)
+ return x
+
+
+def from_int(x: Any) -> int:
+ assert isinstance(x, int) and not isinstance(x, bool)
+ return x
+
+
+class TestTypeMissingFromDict:
+ name: str
+ age: int
+
+ def __init__(self, name: str, age: int) -> None:
+ self.name = name
+ self.age = age
+
+ def to_dict(self) -> dict:
+ result: dict = {}
+ result["name"] = from_str(self.name)
+ result["age"] = from_int(self.age)
+ return result
+
+
+@functions_framework.typed(TestTypeMissingFromDict)
+def function_typed_missing_from_dict(test_type: TestTypeMissingFromDict):
+ valid_event = test_type.name == "john" and test_type.age == 10
+ if not valid_event:
+ raise Exception("Received invalid input")
+ return test_type

tests/test_functions/typed_events/missing_parameter.py

@@ -0,0 +1,23 @@
+# Copyright 2022 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.
+
+"""Function used to test handling functions using typed decorators."""
+import flask
+
+import functions_framework
+
+
+@functions_framework.typed
+def function_typed_missing_type_information():
+ print("hello")