Added use_subprocess to be more generic

tests/test_config.py

@@ -544,3 +544,18 @@ def test_bind_fd_works_with_reload_or_workers(reload, workers):  # pragma: py-wi
     assert sock.getsockname() == ""
     sock.close()
     fdsock.close()
+
+
+@pytest.mark.parametrize(
+    "reload, workers, expected",
+    [
+        (True, 1, True),
+        (False, 2, True),
+        (False, 1, False),
+    ],
+    ids=["--reload=True --workers=1", "--reload=False --workers=2", "--reload=False --workers=1"],
+)
+def test_config_use_subprocess(reload, workers, expected):
+    config = Config(app=asgi_app, reload=reload, workers=workers)
+    config.load()
+    assert config.use_subprocess == expected

uvicorn/config.py

@@ -125,7 +125,7 @@ def create_ssl_context(
     ctx = ssl.SSLContext(ssl_version)
     get_password = (lambda: password) if password else None
     ctx.load_cert_chain(certfile, keyfile, get_password)
-    ctx.verify_mode = cert_reqs
+    ctx.verify_mode = ssl.VerifyMode(cert_reqs)
     if ca_certs:
         ctx.load_verify_locations(ca_certs)
     if ciphers:
@@ -174,7 +174,7 @@ def resolve_reload_patterns(
     for j in range(len(directories)):
         for k in range(j + 1, len(directories)):
             if directories[j] in directories[k].parents:
-                children.append(directories[k])
+                children.append(directories[k])  # pragma: py-darwin
             elif directories[k] in directories[j].parents:
                 children.append(directories[j])
 
@@ -238,7 +238,7 @@ def __init__(
         ssl_cert_reqs: int = ssl.CERT_NONE,
         ssl_ca_certs: Optional[str] = None,
         ssl_ciphers: str = "TLSv1",
-        headers: Optional[List[List[str]]] = None,
+        headers: Optional[List[Tuple[str, str]]] = None,
         factory: bool = False,
     ):
         self.app = app
@@ -280,7 +280,7 @@ def __init__(
         self.ssl_cert_reqs = ssl_cert_reqs
         self.ssl_ca_certs = ssl_ca_certs
         self.ssl_ciphers = ssl_ciphers
-        self.headers: List[List[str]] = headers or []
+        self.headers: List[Tuple[str, str]] = headers or []
         self.encoded_headers: List[Tuple[bytes, bytes]] = []
         self.factory = factory
 
@@ -373,6 +373,10 @@ def asgi_version(self) -> Literal["2.0", "3.0"]:
     def is_ssl(self) -> bool:
         return bool(self.ssl_keyfile or self.ssl_certfile)
 
+    @property
+    def use_subprocess(self) -> bool:
+        return bool(self.reload or self.workers > 1)
+
     def configure_logging(self) -> None:
         logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
 
@@ -503,7 +507,7 @@ def load(self) -> None:
     def setup_event_loop(self) -> None:
         loop_setup: Optional[Callable] = import_from_string(LOOP_SETUPS[self.loop])
         if loop_setup is not None:
-            loop_setup(reload=self.reload, workers=self.workers)
+            loop_setup(use_subprocess=self.use_subprocess)
 
     def bind_socket(self) -> socket.socket:
         logger_args: List[Union[str, int]]

uvicorn/loops/asyncio.py

@@ -1,21 +1,11 @@
 import asyncio
 import logging
 import sys
-from typing import Optional
 
 logger = logging.getLogger("uvicorn.error")
 
 
-def asyncio_setup(
-    reload: bool = False, workers: Optional[int] = None
-) -> None:  # pragma: no cover
-    if (
-        sys.version_info >= (3, 8)
-        and sys.platform == "win32"
-        and any([reload, workers])
-    ):
-        if reload:
-            logger.warning(
-                "The --reload flag should not be used in production on Windows."
-            )
+def asyncio_setup(use_subprocess: bool = False) -> None:  # pragma: no cover
+    if sys.version_info >= (3, 8) and sys.platform == "win32" and use_subprocess:
+        logger.warning("The --reload flag should not be used in production on Windows.")
         asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

uvicorn/loops/auto.py

@@ -1,14 +1,11 @@
-from typing import Optional
-
-
-def auto_loop_setup(reload: bool = False, workers: Optional[int] = None) -> None:
+def auto_loop_setup(use_subprocess: bool = False) -> None:
     try:
         import uvloop  # noqa
     except ImportError:  # pragma: no cover
         from uvicorn.loops.asyncio import asyncio_setup as loop_setup
 
-        loop_setup(reload=reload, workers=workers)
+        loop_setup(use_subprocess=use_subprocess)
     else:  # pragma: no cover
         from uvicorn.loops.uvloop import uvloop_setup
 
-        uvloop_setup(reload=reload, workers=workers)
+        uvloop_setup(use_subprocess=use_subprocess)

uvicorn/loops/uvloop.py

@@ -1,8 +1,7 @@
 import asyncio
-from typing import Optional
 
 import uvloop
 
 
-def uvloop_setup(reload: bool = False, workers: Optional[int] = None) -> None:
+def uvloop_setup(use_subprocess: bool = False) -> None:
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())