fix: revert to separate GunicornApplication and UvicornApplication classes

Remove the BaseGunicornApplication abstraction as it was causing issues with the timeout mechanism. Each class now independently extends gunicorn.app.base.BaseApplication, which is cleaner and avoids the problems we were seeing with shared state and options handling.

Author: taeold

src/functions_framework/_http/gunicorn.py

@@ -27,45 +27,27 @@
 TIMEOUT_SECONDS = None
 
 
-class BaseGunicornApplication(gunicorn.app.base.BaseApplication):
-    """Base Gunicorn application with common configuration."""
+class GunicornApplication(gunicorn.app.base.BaseApplication):
+    """Gunicorn application for WSGI apps with gthread worker support."""
 
     def __init__(self, app, host, port, debug, **options):
         global TIMEOUT_SECONDS
         TIMEOUT_SECONDS = int(os.environ.get("CLOUD_RUN_TIMEOUT_SECONDS", 0))
+        threads = int(os.environ.get("THREADS", (os.cpu_count() or 1) * 4))
 
         self.options = {
             "bind": "%s:%s" % (host, port),
             "workers": int(os.environ.get("WORKERS", 1)),
+            "threads": threads,
+            "timeout": TIMEOUT_SECONDS,
             "loglevel": os.environ.get("GUNICORN_LOG_LEVEL", "error"),
             "limit_request_line": 0,
-            "timeout": TIMEOUT_SECONDS,
         }
-
         self.options.update(options)
         self.app = app
 
         super().__init__()
 
-    def load_config(self):
-        for key, value in self.options.items():
-            self.cfg.set(key, value)
-
-    def load(self):
-        return self.app
-
-
-class GunicornApplication(BaseGunicornApplication):
-    """Gunicorn application for WSGI apps with gthread worker support."""
-
-    def __init__(self, app, host, port, debug, **options):
-        threads = int(os.environ.get("THREADS", (os.cpu_count() or 1) * 4))
-        # Make a copy to avoid mutating the passed-in options dict
-        options_copy = dict(options)
-        options_copy["threads"] = threads
-
-        super().__init__(app, host, port, debug, **options_copy)
-
         # Use custom worker with timeout support if conditions are met
         if (
             TIMEOUT_SECONDS > 0
@@ -78,18 +60,40 @@ def __init__(self, app, host, port, debug, **options):
             # Remove timeout from options when using custom worker
             del self.options["timeout"]
 
+    def load_config(self):
+        for key, value in self.options.items():
+            self.cfg.set(key, value)
+
+    def load(self):
+        return self.app
+
 
 class GThreadWorkerWithTimeoutSupport(ThreadWorker):  # pragma: no cover
     def handle_request(self, req, conn):
         with ThreadingTimeout(TIMEOUT_SECONDS):
             super(GThreadWorkerWithTimeoutSupport, self).handle_request(req, conn)
 
 
-class UvicornApplication(BaseGunicornApplication):
+class UvicornApplication(gunicorn.app.base.BaseApplication):
     """Gunicorn application for ASGI apps using Uvicorn workers."""
 
     def __init__(self, app, host, port, debug, **options):
-        # Make a copy to avoid mutating the passed-in options dict
-        options_copy = dict(options)
-        options_copy["worker_class"] = "uvicorn_worker.UvicornWorker"
-        super().__init__(app, host, port, debug, **options_copy)
+        self.options = {
+            "bind": "%s:%s" % (host, port),
+            "workers": int(os.environ.get("WORKERS", 1)),
+            "worker_class": "uvicorn_worker.UvicornWorker",
+            "timeout": int(os.environ.get("CLOUD_RUN_TIMEOUT_SECONDS", 0)),
+            "loglevel": os.environ.get("GUNICORN_LOG_LEVEL", "error"),
+            "limit_request_line": 0,
+        }
+        self.options.update(options)
+        self.app = app
+
+        super().__init__()
+
+    def load_config(self):
+        for key, value in self.options.items():
+            self.cfg.set(key, value)
+
+    def load(self):
+        return self.app