Allow worker kwargs

tests/test_config.py

@@ -54,6 +54,14 @@ def wsgi_app(environ: Environ, start_response: StartResponse) -> None:
     pass  # pragma: nocover
 
 
+def asgi_app_factory(**kwargs):
+    async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+        pass  # pragma: nocover
+
+    app.worker_kwargs = kwargs  # type: ignore
+    return app
+
+
 @pytest.mark.parametrize(
     "app, expected_should_reload",
     [(asgi_app, False), ("tests.test_config:asgi_app", True)],
@@ -545,3 +553,16 @@ def test_warn_when_using_reload_and_workers(caplog: pytest.LogCaptureFixture) ->
     Config(app=asgi_app, reload=True, workers=2)
     assert len(caplog.records) == 1
     assert '"workers" flag is ignored when reloading is enabled.' in caplog.records[0].message
+
+
+def test_config_worker_kwargs():
+    config = Config(
+        app="tests.test_config:asgi_app_factory",
+        worker_kwargs={"alpha": 12, "beta": [3, 4, 5]},
+        factory=True,
+        workers=4,
+        proxy_headers=False,
+    )
+    config.load()
+
+    assert config.loaded_app.worker_kwargs == {"alpha": 12, "beta": [3, 4, 5]}

uvicorn/config.py

@@ -223,6 +223,7 @@ def __init__(
         headers: list[tuple[str, str]] | None = None,
         factory: bool = False,
         h11_max_incomplete_event_size: int | None = None,
+        worker_kwargs: dict | None = None,
     ):
         self.app = app
         self.host = host
@@ -268,6 +269,7 @@ def __init__(
         self.encoded_headers: list[tuple[bytes, bytes]] = []
         self.factory = factory
         self.h11_max_incomplete_event_size = h11_max_incomplete_event_size
+        self.worker_kwargs = worker_kwargs or {}
 
         self.loaded = False
         self.configure_logging()
@@ -437,7 +439,7 @@ def load(self) -> None:
             sys.exit(1)
 
         try:
-            self.loaded_app = self.loaded_app()
+            self.loaded_app = self.loaded_app(**self.worker_kwargs)
         except TypeError as exc:
             if self.factory:
                 logger.error("Error loading ASGI app factory: %s", exc)

uvicorn/main.py

@@ -507,6 +507,7 @@ def run(
     app_dir: str | None = None,
     factory: bool = False,
     h11_max_incomplete_event_size: int | None = None,
+    worker_kwargs: dict | None = None,
 ) -> None:
     if app_dir is not None:
         sys.path.insert(0, app_dir)
@@ -558,6 +559,7 @@ def run(
         use_colors=use_colors,
         factory=factory,
         h11_max_incomplete_event_size=h11_max_incomplete_event_size,
+        worker_kwargs=worker_kwargs,
     )
     server = Server(config=config)