Fixes GoogleCloudPlatform#174.

I've added ability to configure Gunicorn behavior using environment variables.

src/functions_framework/_http/gunicorn.py

@@ -11,9 +11,51 @@
 # 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 os
+from typing import Any, Optional
 
 import gunicorn.app.base
 
+GUNICORN_OPTIONS_SEPARATOR_ENV: str = "GUNICORN_OPTIONS_SEPARATOR"
+"""The name of the env variable that holds a separator of the Gunicorn options."""
+GUNICORN_OPTIONS_ENV: str = "GUNICORN_OPTIONS"
+"""The name of the env variable that holds Gunicorn options in a key=value format
+where each option is separated from the other one with a GUNICORN_OPTIONS_SEPARATOR 
+(or `,`) by default.
+"""
+
+
+def _gunicorn_env_options() -> dict[str, Any]:
+    """Parses Gunicorn options provided through environment variable if any are provided.
+
+    The Gunicorn options are specified using `GUNICORN_OPTIONS_<option-name>` formatted options
+    that can override the standard provided options if specified.
+    """
+    gunicorn_options: Optional[str] = os.getenv(GUNICORN_OPTIONS_ENV)
+    if not gunicorn_options:
+        return {}
+    options_separator: str = os.getenv(GUNICORN_OPTIONS_SEPARATOR_ENV, ",")
+    options: list[str] = gunicorn_options.split(options_separator)
+    result = {}
+    for option in options:
+        option_config = option.split("=", maxsplit=2)
+        if len(option_config) == 2:
+            key, value = option_config
+        elif len(option_config) == 3:
+            key, value, value_type = option_config
+            if value_type == "int":
+                value = int(value)
+            elif value_type == "float":
+                value = float(value)
+            elif value_type == "bool":
+                value = bool(value)
+        else:
+            raise TypeError(
+                f"Gunicorn option config must be of format key=value(=type), but was: {option}"
+            )
+        result[key] = value
+    return result
+
 
 class GunicornApplication(gunicorn.app.base.BaseApplication):
     def __init__(self, app, host, port, debug, **options):
@@ -26,6 +68,7 @@ def __init__(self, app, host, port, debug, **options):
             "limit_request_line": 0,
         }
         self.options.update(options)
+        self.options.update(_gunicorn_env_options())
         self.app = app
         super().__init__()
 

tests/test_http.py

@@ -11,7 +11,7 @@
 # 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 os
 import platform
 import sys
 
@@ -110,6 +110,46 @@ def test_gunicorn_application(debug):
     assert gunicorn_app.load() == app
 
 
+@pytest.mark.skipif("platform.system() == 'Windows'")
+@pytest.mark.parametrize("workers, threads, keepalive", [(1, 4, 50), (2, 8, 1000)])
+def test_gunicorn_application_custom_options(workers, threads, keepalive):
+    app = pretend.stub()
+    host = "1.2.3.4"
+    port = "1234"
+    options = {}
+
+    import functions_framework._http.gunicorn
+
+    os.environ[functions_framework._http.gunicorn.GUNICORN_OPTIONS_SEPARATOR_ENV] = "|"
+    os.environ[functions_framework._http.gunicorn.GUNICORN_OPTIONS_ENV] = (
+        f"workers={workers}=int|threads={threads}=int|keepalive={keepalive}=int"
+    )
+
+    gunicorn_app = functions_framework._http.gunicorn.GunicornApplication(
+        app, host, port, False, **options
+    )
+    os.environ.pop(functions_framework._http.gunicorn.GUNICORN_OPTIONS_SEPARATOR_ENV)
+    os.environ.pop(functions_framework._http.gunicorn.GUNICORN_OPTIONS_ENV)
+
+    assert gunicorn_app.app == app
+    assert gunicorn_app.options == {
+        "bind": "%s:%s" % (host, port),
+        "workers": workers,
+        "threads": threads,
+        "timeout": 0,
+        "loglevel": "error",
+        "limit_request_line": 0,
+        "keepalive": keepalive
+    }
+
+    assert gunicorn_app.cfg.bind == ["1.2.3.4:1234"]
+    assert gunicorn_app.cfg.workers == workers
+    assert gunicorn_app.cfg.threads == threads
+    assert gunicorn_app.cfg.timeout == 0
+    assert gunicorn_app.cfg.keepalive == keepalive
+    assert gunicorn_app.load() == app
+
+
 @pytest.mark.parametrize("debug", [True, False])
 def test_flask_application(debug):
     app = pretend.stub(run=pretend.call_recorder(lambda *a, **kw: None))