fabiommendes/sidekick

View on GitHub
sidekick-beta/sidekick/beta/app/gcf.py

Summary

Maintainability
A
1 hr
Test Coverage
import json
import os
from functools import lru_cache
from pathlib import Path
from typing import TYPE_CHECKING, Dict, Any, Tuple

from .cli import main as _main, parse_value
from .function_wrapper import wrap_function
from ...functions import signature
from ...proxy import import_later


# if TYPE_CHECKING or 1:
import flask
import functions_framework as runtime
from functions_framework import _http as server
import jsonrpc

# else:
#     jsonrpc = import_later("jsonrpc", pypi="sidekick[app.gcf]")
#     runtime = import_later("functions_framework", pypi="sidekick[app.gcf]")
#     server = import_later("functions_framework._http", pypi="sidekick[app.gcf]")
#     flask = import_later("flask", pypi="sidekick[app.gcf]")

DEFAULT_HANDLERS = {}
HTTP_METHODS = ("GET", "POST", "PATCH")
register_handler = lambda key: lambda fn: DEFAULT_HANDLERS.setdefault(key, fn)


def run(
    func: callable,
    signature_type="http",
    host="0.0.0.0",
    port=8080,
    debug=True,
    path=None,
    handlers=None,
    methods=HTTP_METHODS,
):
    """
    Run function in a "functions_framework" runtime. This emulates the GCF
    emulator locally.
    """

    methods = HTTP_METHODS if methods is None else methods
    handler_methods = bind_handlers(func, handlers, methods)

    def cloud_function(request: "flask.Request"):
        handler = get_handler(request, handler_methods)
        return handler(request, handler_methods)

    app = _create_app(cloud_function, signature_type=signature_type, path=path)
    server.create_server(app, debug).run(host, port)


def deploy(func: callable):
    """
    Deploy function to Google Cloud Platform.
    """


#
# Wrap function
#
def bind_handlers(func_or_mod, handlers, methods):
    """
    Receive a function and a mapping from {(mime, method): handler_factory}
    and return a populated mapping of handlers.
    """
    if handlers is None:
        handlers = DEFAULT_HANDLERS

    res = {k: mk_handler(func_or_mod) for k, mk_handler in handlers.items()}
    for k, v in list(res.items()):
        if isinstance(k, str):
            res.update({(k, m): v for m in methods if (k, m) not in res})

    if None in handlers:
        default_handler = handlers[None]
        for m in methods:
            res.setdefault((None, m), default_handler)

    return res


def get_handler(request: "flask.Request", handlers: Dict[Tuple[str, str], callable]):
    """
    Get handler from request.

    Handler is a function that receives a request and return a response.
    """
    method = request.method
    for mime in request.accept_mimetypes:
        try:
            fn = handlers[mime[0], method]
            print(mime[0], method)
            return fn
        except KeyError:
            pass
    try:
        return handlers[None, method]
    except KeyError:
        msg = f"Invalid method/mime: {request.method}, {request.accept_mimetypes}"
        return lambda *_: flask.Response(msg, 500)


@register_handler("application/json")
def jsonrpc_handler(func, info=True):
    dispatcher = jsonrpc_dispatcher(func, info)

    def handler(request, _):
        response = jsonrpc.JSONRPCResponseManager.handle(request.data, dispatcher)
        return flask.Response(response.json, mimetype="application/json")

    return handler


@register_handler(("application/json", "GET"))
def jsonrpc_handler_GET(func, info=True):
    dispatcher = jsonrpc_dispatcher(func, info)

    def handler(request: "flask.Request", _):
        params = dict(request.args)
        rpc_id = params.pop("json-rpc-id", 0)
        payload = {
            "jsonrpc": "2.0",
            "method": request.path.strip("/"),
            "params": {k: parse_value(v) for k, v in params.items()},
            "id": rpc_id,
        }
        request = jsonrpc.jsonrpc.JSONRPCRequest.from_data(payload)
        response = jsonrpc.JSONRPCResponseManager.handle_request(request, dispatcher)
        data = f"<h1>JSON response</h1><pre>{json.dumps(response.data, indent=2)}</pre>"
        return flask.Response(data, mimetype="text/html")

    return handler


@lru_cache(32)
def jsonrpc_dispatcher(func, info=True) -> "jsonrpc.Dispatcher":
    dispatcher = jsonrpc.Dispatcher()
    if callable(func):
        dispatcher[func.__name__] = wrap_function(func)
    else:
        raise NotImplementedError

    if info:
        info = lambda: {k: spec(fn) for k, fn in dispatcher.items()}
        dispatcher.setdefault("info", lru_cache(1)(info))

    return dispatcher


@register_handler("text/html")
def html_handler(func):
    """
    Generic handler that redirects to JSON.
    """

    def handler(request, handlers):
        fn = handlers["application/json", request.method]
        return fn(request, handlers)

    return handler


def arguments(request: "flask.Request", func) -> Dict[str, Any]:
    """
    Read arguments from request and return a mapping that will be passed to
    function.
    """
    return {k: parse_value(v) for k, v in request.values.items()}


def response(request: "flask.Request", result: Any) -> "flask.Response":
    """
    Compute flask Response for the given value.
    """
    return flask.make_response(str(result))


@lru_cache(64)
def spec(func):
    """
    A JSON-like element with function specification.
    """
    # TODO: convert to valid RPC-JSON.
    sig = signature(func)
    return {
        "name": func.__name__,
        "args": dict(zip(sig.argnames(), map(repr_type, sig.args()))),
    }


@lru_cache(64)
def repr_type(t):
    """
    Convert type to string representation.
    """
    return str(t)


#
# This was copied and adapted from functions_framework
#
def _create_app(function, signature_type=None, path=None):
    path = Path(path or os.getcwd())
    template_folder = str(path / "templates")

    # Set the environment variable if it wasn't already
    os.environ["FUNCTION_SIGNATURE_TYPE"] = signature_type
    os.environ["TARGET"] = target = function.__name__
    os.environ["SOURCE"] = "<string>"

    app = flask.Flask(target, template_folder=template_folder)
    app.config["MAX_CONTENT_LENGTH"] = runtime.MAX_CONTENT_LENGTH

    # 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
    import werkzeug

    if signature_type == "http":
        app.url_map.add(
            werkzeug.routing.Rule("/", defaults={"path": ""}, endpoint="run")
        )
        app.url_map.add(werkzeug.routing.Rule("/robots.txt", endpoint="error"))
        app.url_map.add(werkzeug.routing.Rule("/favicon.ico", endpoint="error"))
        app.url_map.add(werkzeug.routing.Rule("/<path:path>", endpoint="run"))
        app.view_functions["run"] = runtime._http_view_func_wrapper(
            function, flask.request
        )
        app.view_functions["error"] = lambda: flask.abort(404, description="Not Found")
        app.after_request(runtime.read_request)
    elif signature_type == "event" or signature_type == "cloudevent":
        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"]
            )
        )

        # Add a dummy endpoint for GET /
        app.url_map.add(werkzeug.routing.Rule("/", endpoint="get", methods=["GET"]))
        app.view_functions["get"] = lambda: ""

        # Add the view functions
        app.view_functions["event"] = runtime._event_view_func_wrapper(
            function, flask.request
        )
        app.view_functions["cloudevent"] = runtime._cloudevent_view_func_wrapper(
            function, flask.request
        )
    else:
        raise runtime.FunctionsFrameworkException(
            "Invalid signature type: {signature_type}".format(
                signature_type=signature_type
            )
        )

    return app


if __name__ == "__main__":
    _main(run)