Gunicorn settings class (#16)

* Add GunicornSettings class

- Uses Pydantic BaseSettings class

* Add tests for gunicorn conf

* Minor tweaks to gunicorn module:
- modify module docstring
- remove pylint rule disable since it's not needed
- add link to gunicorn settings docs

Author: grahamalama

infra/gunicorn_conf.py

@@ -1,58 +1,63 @@
 """
-Conf file extracted from:
+Conf file adapted from:
 https://github.com/tiangolo/uvicorn-gunicorn-docker/blob/2daa3e3873c837d5781feb4ff6a40a89f791f81b/docker-images/gunicorn_conf.py
 """
-# pylint: disable=invalid-name
 
 import multiprocessing
-import os
+from typing import Optional
 
 from prometheus_client import multiprocess
+from pydantic import BaseSettings, conint, root_validator, validator
 
-workers_per_core_str = os.getenv("WORKERS_PER_CORE", "1")
-max_workers_str = os.getenv("MAX_WORKERS")
-use_max_workers = None
-if max_workers_str:
-    use_max_workers = int(max_workers_str)
-web_concurrency_str = os.getenv("WEB_CONCURRENCY", None)
 
-host = os.getenv("HOST", "0.0.0.0")
-port = os.getenv("PORT", "8000")
-bind_env = os.getenv("BIND", None)
-use_loglevel = os.getenv("LOG_LEVEL", "info")
-if bind_env:
-    use_bind = bind_env
-else:
-    use_bind = f"{host}:{port}"
+class GunicornSettings(BaseSettings):
+    keep_alive: int = 5
+    timeout: int = 120
+    graceful_timeout: int = 120
+    log_level: str = "info"
+    host: str = "0.0.0.0"
+    port: str = "8000"
+    bind: Optional[str]
+    access_log: str = "-"
+    error_log: str = "-"
+    workers_per_core: float = 1.0
+    max_workers: Optional[int]
+    web_concurrency: Optional[conint(gt=0)]
+    workers: Optional[int]
 
-cores = multiprocessing.cpu_count()
-workers_per_core = float(workers_per_core_str)
-default_web_concurrency = workers_per_core * cores
-if web_concurrency_str:
-    web_concurrency = int(web_concurrency_str)
-    assert web_concurrency > 0
-else:
-    web_concurrency = max(int(default_web_concurrency), 2)
-    if use_max_workers:
-        web_concurrency = min(web_concurrency, use_max_workers)
-accesslog_var = os.getenv("ACCESS_LOG", "-")
-use_accesslog = accesslog_var or None
-errorlog_var = os.getenv("ERROR_LOG", "-")
-use_errorlog = errorlog_var or None
-graceful_timeout_str = os.getenv("GRACEFUL_TIMEOUT", "120")
-timeout_str = os.getenv("TIMEOUT", "120")
-keepalive_str = os.getenv("KEEP_ALIVE", "5")
+    @validator("bind")
+    def set_bind(cls, bind, values):
+        return bind if bind else f"{values['host']}:{values['port']}"
+
+    @root_validator(skip_on_failure=True)
+    def set_workers(cls, values):
+        if values["workers"] is not None:
+            return values
+        elif values["web_concurrency"]:
+            values["workers"] = values["web_concurrency"]
+        else:
+            cores = multiprocessing.cpu_count()
+            default_workers = values["workers_per_core"] * cores
+            workers = max(int(default_workers), 2)
+            if values["max_workers"]:
+                workers = min(workers, values["max_workers"])
+            values["workers"] = workers
+        return values
+
+
+gunicorn_settings = GunicornSettings()
 
 # Gunicorn config variables
-loglevel = use_loglevel
-workers = web_concurrency
-bind = use_bind
-errorlog = use_errorlog
+# https://docs.gunicorn.org/en/stable/settings.html#settings
+loglevel = gunicorn_settings.log_level
+workers = gunicorn_settings.workers
+bind = gunicorn_settings.bind
+errorlog = gunicorn_settings.error_log
 worker_tmp_dir = "/dev/shm"
-accesslog = use_accesslog
-graceful_timeout = int(graceful_timeout_str)
-timeout = int(timeout_str)
-keepalive = int(keepalive_str)
+accesslog = gunicorn_settings.access_log
+graceful_timeout = gunicorn_settings.graceful_timeout
+timeout = gunicorn_settings.timeout
+keepalive = gunicorn_settings.keep_alive
 
 
 # For debugging and testing
@@ -66,10 +71,10 @@
     "errorlog": errorlog,
     "accesslog": accesslog,
     # Additional, non-gunicorn variables
-    "workers_per_core": workers_per_core,
-    "use_max_workers": use_max_workers,
-    "host": host,
-    "port": port,
+    "workers_per_core": gunicorn_settings.workers_per_core,
+    "use_max_workers": gunicorn_settings.max_workers,
+    "host": gunicorn_settings.host,
+    "port": gunicorn_settings.port,
 }
 
 

tests/unit/test_gunicorn_conf.py

@@ -0,0 +1,43 @@
+from pydantic import ValidationError
+import pytest
+from infra.gunicorn_conf import GunicornSettings
+
+
+def test_set_bind(monkeypatch):
+    expected = "1.2.3.4:5678"
+    monkeypatch.setenv("BIND", expected)
+    settings = GunicornSettings()
+    assert settings.bind == expected
+
+
+class TestSetWorkers:
+    def test_workers_env_var(self, monkeypatch):
+        expected = 3
+        monkeypatch.setenv("WORKERS", str(expected))
+        settings = GunicornSettings()
+        assert settings.workers == expected
+
+    def test_web_concurency_env_var(self, monkeypatch):
+        expected = 3
+        monkeypatch.setenv("WEB_CONCURRENCY", str(expected))
+        settings = GunicornSettings()
+        assert settings.workers == expected
+
+    def test_web_concurency_env_var_out_of_range(self, monkeypatch):
+        expected = 0
+        monkeypatch.setenv("WEB_CONCURRENCY", str(expected))
+        with pytest.raises(ValidationError):
+            GunicornSettings()
+
+    def test_default_workers(self, monkeypatch):
+        expected = 2
+        monkeypatch.setattr("multiprocessing.cpu_count", lambda: 2)
+        settings = GunicornSettings()
+        assert settings.workers == expected
+
+    def test_max_workers(self, monkeypatch):
+        expected = 1
+        monkeypatch.setattr("multiprocessing.cpu_count", lambda: 2)
+        monkeypatch.setenv("MAX_WORKERS", str(expected))
+        settings = GunicornSettings()
+        assert settings.workers == expected