def create_app()

in src/functions_framework/__init__.py [0:0]

• 47 lines of code
• 8 McCabe index (conditional complexity)

def create_app(target=None, source=None, signature_type=None):
    target = _function_registry.get_function_target(target)
    source = _function_registry.get_function_source(source)

    # Set the template folder relative to the source path
    # Python 3.5: join does not support PosixPath
    template_folder = str(pathlib.Path(source).parent / "templates")

    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)

    if _enable_execution_id_logging():
        _configure_app_execution_id_logging()

    # Create the application
    _app = flask.Flask(target, template_folder=template_folder)
    _app.register_error_handler(500, crash_handler)
    global errorhandler
    errorhandler = _app.errorhandler

    # Handle legacy GCF Python 3.7 behavior
    if os.environ.get("ENTRY_POINT"):
        os.environ["FUNCTION_NAME"] = os.environ.get("K_SERVICE", target)
        _app.make_response_original = _app.make_response

        def handle_none(rv):
            if rv is None:
                rv = "OK"
            return _app.make_response_original(rv)

        _app.make_response = handle_none

        # Handle log severity backwards compatibility
        sys.stdout = _LoggingHandler("INFO", sys.stderr)
        sys.stderr = _LoggingHandler("ERROR", sys.stderr)
        setup_logging()

    _app.wsgi_app = execution_id.WsgiMiddleware(_app.wsgi_app)
    # Execute the module, within the application context
    with _app.app_context():
        try:
            spec.loader.exec_module(source_module)
            function = _function_registry.get_user_function(
                source, source_module, target
            )
        except Exception as e:
            if werkzeug.serving.is_running_from_reloader():
                # When reloading, print out the error immediately, but raise
                # it later so the debugger or server can handle it.
                import traceback

                traceback.print_exc()
                err = e

                def function(*_args, **_kwargs):
                    raise err from None

            else:
                # When not reloading, raise the error immediately so the
                # command fails.
                raise e from None

    # Get the configured function signature type
    signature_type = _function_registry.get_func_signature_type(target, signature_type)

    _configure_app(_app, function, signature_type)

    return _app