Type-annotate uvicorn/config.py (#1067)

* Add changes from PR #992 This commit will bring in changes to uvicorn/config.py added by @Kludex in PR #992, updating for the latest master branch. * Correct SSL type annotations #992 https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_cert_chain - `certfile` is a required positional argument when running `SSLContext.load_cert_chain`, so annotating as `Optional` (which allows `None`) would not be ideal. Path-like objects are acceptable, so after `from pathlib import Path`, the annotation is `certfile: Union[Path, str]`. - `if self.is_ssl and self.ssl_certfile` will help ensure that the `self.ssl_certfile` required positional argument is present. * Simplify log_config type annotation `Dict[str, Any]` can be simplified to `dict`. * Add type annotation for app #990 #992 app is sometimes a string, as described in uvicorn/main.py. In other cases, especially in the tests, app is an ASGI application callable. This commit will add a type annotation for the app argument to address these use cases. * Simplify if expression in Config().headers * Allow paths to be used for Config(env_file) * Add asyncio Protocol types to class Config kwargs Protocol classes are sometimes used for the app kwarg, such as in tests/test_config.py. * Improve use of Literal type for ASGI interfaces https://mypy.readthedocs.io/en/stable/literal_types.html This commit will retain the use of literals to define ASGI protocol versions, but improve and correct the use of literal types. As described in the mypy docs, `Literal["2.0", "3.0"]` is a simpler way to write `Union[Literal["2.0"], Literal["3.0"]]`. * Add type annotation to Config kwargs in workers.py https://mypy.readthedocs.io/en/stable/type_inference_and_annotations.html This is just one of those times when mypy needs a little help. * Add type annotations to test_config.py PR #978 added logic to convert `reload_dirs` from string to list, as shown in `test_reload_dir_is_set()`, so the annotation on reload_dirs will be updated to accept a string. * Type-annotate constants in uvicorn/config.py Prevents mypy `[has-type]` errors ("Cannot determine type of {object}") * Correct type annotations in uvicorn/importer.py #1046 #1067 The relationship between module imports and calls in uvicorn/config.py was previously unclear. `import_from_string` was annotated as returning a `ModuleType`, but was being used as if it were returning callables. Mypy was raising "module not callable" errors, as reported in #1046. Current use cases of `import_from_string` include: - `uvicorn.Config.HTTP_PROTOCOLS`, `uvicorn.Config.WS_PROTOCOLS`: `Type[asyncio.Protocol]` (these are subclasses of asyncio protocols, so they are annotated with `Type[Class]` as explained in the mypy docs - `uvicorn.Config.LOOP_SETUPS`: `Callable` (each loop is a function) - `uvicorn.Config().loaded_app`: `ASGIApplication` (should be an ASGI application instance, like `Starlette()` or `FastAPI()`) Ideally, the return type of `import_from_string` would reflect these use cases, but the complex typing will be difficult to maintain. For easier maintenance, the return type will be `Any`, and objects returned by `import_from_string` will be annotated directly. * Use os.PathLike for paths in uvicorn/config.py #1067 (comment) Alternative to `pathlib.Path` introduced in Python 3.6. * Use more specific types in test_config.py #1067 (comment) #1067 (comment) https://github.com/encode/starlette/blob/b6f3578bb2cf6c60e3efe110143409b47f368d36/starlette/config.py#L16 https://github.com/python/cpython/blob/3e1c7167d86a2a928cdcb659094aa10bb5550c4c/Lib/os.py#L737 https://docs.pytest.org/en/latest/reference/reference.html#pytest.FixtureRequest - Add custom types for exceptions and start response - Use `MutableMapping` for `os.environ`: matches starlette/config.py. - Use `pytest.FixtureRequest` for fixture requests. The `param` attribute is optional, so mypy requires a check for the attribute before indexing into it (`getattr(request, "param")`). * Install missing YAML type stubs for mypy #1067 Fixes `[import]` error: Library stubs not installed for "yaml" * Add Environ type to test_config.py #1067 (comment) * Add Literal type aliases for web server config #1067 (comment) * Use suggested casing for Literal type aliases #1067 (comment) #1067 (comment) * Restore test_config.py test_app_factory comment #1067 (comment) Incorrectly modified in c436bba. * Simplify event loop setup in config.py #1067 (comment) * Remove old type comment after merging master #1067 (comment) * Assert that certfile is present for SSL context #1067 (comment) https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_cert_chain `certfile` is a required argument for `SSLContext.load_cert_chain`. As it is currently, `is_ssl` could return `True` without `certfile`. * Restore support for Config(loop='none') #455 e9e59e9 e60ba66 #1067 (comment) * Move WSGI types to uvicorn/_types.py #1067 (comment) * Remove Awaitable from app type annotation #1067 (comment) Co-authored-by: euri10 <euri10@users.noreply.github.com>
encode · Nov 17, 2021 · 6155cab · 6155cab
1 parent 69273a0
commit 6155cab
Show file tree

Hide file tree

Showing 8 changed files with 189 additions and 119 deletions.
diff --git a/requirements.txt b/requirements.txt
@@ -16,6 +16,7 @@ pytest
 pytest-mock
 mypy
 types-click
+types-pyyaml
 trustme
 cryptography
 coverage

diff --git a/setup.cfg b/setup.cfg
@@ -9,9 +9,12 @@ follow_imports = silent
 files =
     uvicorn/lifespan,
     tests/test_lifespan.py,
+    uvicorn/config.py,
+    tests/test_config.py,
     uvicorn/middleware/message_logger.py,
     uvicorn/supervisors/basereload.py,
     uvicorn/importer.py,
+    tests/importer/test_importer.py,
     uvicorn/protocols/utils.py,
     uvicorn/loops,
     uvicorn/main.py,

diff --git a/tests/importer/test_importer.py b/tests/importer/test_importer.py
@@ -3,40 +3,40 @@
 from uvicorn.importer import ImportFromStringError, import_from_string
 
 
-def test_invalid_format():
+def test_invalid_format() -> None:
     with pytest.raises(ImportFromStringError) as exc_info:
         import_from_string("example:")
     expected = 'Import string "example:" must be in format "<module>:<attribute>".'
     assert expected in str(exc_info.value)
 
 
-def test_invalid_module():
+def test_invalid_module() -> None:
     with pytest.raises(ImportFromStringError) as exc_info:
         import_from_string("module_does_not_exist:myattr")
     expected = 'Could not import module "module_does_not_exist".'
     assert expected in str(exc_info.value)
 
 
-def test_invalid_attr():
+def test_invalid_attr() -> None:
     with pytest.raises(ImportFromStringError) as exc_info:
         import_from_string("tempfile:attr_does_not_exist")
     expected = 'Attribute "attr_does_not_exist" not found in module "tempfile".'
     assert expected in str(exc_info.value)
 
 
-def test_internal_import_error():
+def test_internal_import_error() -> None:
     with pytest.raises(ImportError):
         import_from_string("tests.importer.raise_import_error:myattr")
 
 
-def test_valid_import():
+def test_valid_import() -> None:
     instance = import_from_string("tempfile:TemporaryFile")
     from tempfile import TemporaryFile
 
     assert instance == TemporaryFile
 
 
-def test_no_import_needed():
+def test_no_import_needed() -> None:
     from tempfile import TemporaryFile
 
     instance = import_from_string(TemporaryFile)

diff --git a/tests/test_config.py b/tests/test_config.py
@@ -2,11 +2,23 @@
 import logging
 import os
 import socket
+import sys
+import typing
 from copy import deepcopy
+from pathlib import Path
+from unittest.mock import MagicMock
+
+if sys.version_info < (3, 8):
+    from typing_extensions import Literal
+else:
+    from typing import Literal
 
 import pytest
 import yaml
+from asgiref.typing import ASGIApplication, ASGIReceiveCallable, ASGISendCallable, Scope
+from pytest_mock import MockerFixture
 
+from uvicorn._types import Environ, StartResponse
 from uvicorn.config import LOGGING_CONFIG, Config
 from uvicorn.middleware.debug import DebugMiddleware
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
@@ -15,34 +27,36 @@
 
 
 @pytest.fixture
-def mocked_logging_config_module(mocker):
+def mocked_logging_config_module(mocker: MockerFixture) -> MagicMock:
     return mocker.patch("logging.config")
 
 
 @pytest.fixture(scope="function")
-def logging_config():
+def logging_config() -> dict:
     return deepcopy(LOGGING_CONFIG)
 
 
 @pytest.fixture
-def json_logging_config(logging_config):
+def json_logging_config(logging_config: dict) -> str:
     return json.dumps(logging_config)
 
 
 @pytest.fixture
-def yaml_logging_config(logging_config):
+def yaml_logging_config(logging_config: dict) -> str:
     return yaml.dump(logging_config)
 
 
-async def asgi_app(scope, receive, send):
+async def asgi_app(
+    scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+) -> None:
     pass  # pragma: nocover
 
 
-def wsgi_app(environ, start_response):
+def wsgi_app(environ: Environ, start_response: StartResponse) -> None:
     pass  # pragma: nocover
 
 
-def test_debug_app():
+def test_debug_app() -> None:
     config = Config(app=asgi_app, debug=True, proxy_headers=False)
     config.load()
 
@@ -54,7 +68,9 @@ def test_debug_app():
     "app, expected_should_reload",
     [(asgi_app, False), ("tests.test_config:asgi_app", True)],
 )
-def test_config_should_reload_is_set(app, expected_should_reload):
+def test_config_should_reload_is_set(
+    app: ASGIApplication, expected_should_reload: bool
+) -> None:
     config_debug = Config(app=app, debug=True)
     assert config_debug.debug is True
     assert config_debug.should_reload is expected_should_reload
@@ -64,12 +80,12 @@ def test_config_should_reload_is_set(app, expected_should_reload):
     assert config_reload.should_reload is expected_should_reload
 
 
-def test_reload_dir_is_set():
+def test_reload_dir_is_set() -> None:
     config = Config(app=asgi_app, reload=True, reload_dirs="reload_me")
     assert config.reload_dirs == ["reload_me"]
 
 
-def test_wsgi_app():
+def test_wsgi_app() -> None:
     config = Config(app=wsgi_app, interface="wsgi", proxy_headers=False)
     config.load()
 
@@ -78,21 +94,21 @@ def test_wsgi_app():
     assert config.asgi_version == "3.0"
 
 
-def test_proxy_headers():
+def test_proxy_headers() -> None:
     config = Config(app=asgi_app)
     config.load()
 
     assert config.proxy_headers is True
     assert isinstance(config.loaded_app, ProxyHeadersMiddleware)
 
 
-def test_app_unimportable_module():
+def test_app_unimportable_module() -> None:
     config = Config(app="no.such:app")
     with pytest.raises(ImportError):
         config.load()
 
 
-def test_app_unimportable_other(caplog):
+def test_app_unimportable_other(caplog: pytest.LogCaptureFixture) -> None:
     config = Config(app="tests.test_config:app")
     with pytest.raises(SystemExit):
         config.load()
@@ -107,8 +123,8 @@ def test_app_unimportable_other(caplog):
     )
 
 
-def test_app_factory(caplog):
-    def create_app():
+def test_app_factory(caplog: pytest.LogCaptureFixture) -> None:
+    def create_app() -> ASGIApplication:
         return asgi_app
 
     config = Config(app=create_app, factory=True, proxy_headers=False)
@@ -131,21 +147,24 @@ def create_app():
         config.load()
 
 
-def test_concrete_http_class():
+def test_concrete_http_class() -> None:
     config = Config(app=asgi_app, http=H11Protocol)
     config.load()
     assert config.http_protocol_class is H11Protocol
 
 
-def test_socket_bind():
+def test_socket_bind() -> None:
     config = Config(app=asgi_app)
     config.load()
     sock = config.bind_socket()
     assert isinstance(sock, socket.socket)
     sock.close()
 
 
-def test_ssl_config(tls_ca_certificate_pem_path, tls_ca_certificate_private_key_path):
+def test_ssl_config(
+    tls_ca_certificate_pem_path: str,
+    tls_ca_certificate_private_key_path: str,
+) -> None:
     config = Config(
         app=asgi_app,
         ssl_certfile=tls_ca_certificate_pem_path,
@@ -156,7 +175,7 @@ def test_ssl_config(tls_ca_certificate_pem_path, tls_ca_certificate_private_key_
     assert config.is_ssl is True
 
 
-def test_ssl_config_combined(tls_certificate_pem_path):
+def test_ssl_config_combined(tls_certificate_pem_path: str) -> None:
     config = Config(
         app=asgi_app,
         ssl_certfile=tls_certificate_pem_path,
@@ -166,8 +185,10 @@ def test_ssl_config_combined(tls_certificate_pem_path):
     assert config.is_ssl is True
 
 
-def asgi2_app(scope):
-    async def asgi(receive, send):  # pragma: nocover
+def asgi2_app(scope: Scope) -> typing.Callable:
+    async def asgi(
+        receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:  # pragma: nocover
         pass
 
     return asgi  # pragma: nocover
@@ -176,7 +197,9 @@ async def asgi(receive, send):  # pragma: nocover
 @pytest.mark.parametrize(
     "app, expected_interface", [(asgi_app, "3.0"), (asgi2_app, "2.0")]
 )
-def test_asgi_version(app, expected_interface):
+def test_asgi_version(
+    app: ASGIApplication, expected_interface: Literal["2.0", "3.0"]
+) -> None:
     config = Config(app=app)
     config.load()
     assert config.asgi_version == expected_interface
@@ -191,7 +214,11 @@ def test_asgi_version(app, expected_interface):
         pytest.param(False, False, id="use_colors_disabled"),
     ],
 )
-def test_log_config_default(mocked_logging_config_module, use_colors, expected):
+def test_log_config_default(
+    mocked_logging_config_module: MagicMock,
+    use_colors: typing.Optional[bool],
+    expected: typing.Optional[bool],
+) -> None:
     """
     Test that one can specify the use_colors option when using the default logging
     config.
@@ -206,8 +233,11 @@ def test_log_config_default(mocked_logging_config_module, use_colors, expected):
 
 
 def test_log_config_json(
-    mocked_logging_config_module, logging_config, json_logging_config, mocker
-):
+    mocked_logging_config_module: MagicMock,
+    logging_config: dict,
+    json_logging_config: str,
+    mocker: MockerFixture,
+) -> None:
     """
     Test that one can load a json config from disk.
     """
@@ -224,12 +254,12 @@ def test_log_config_json(
 
 @pytest.mark.parametrize("config_filename", ["log_config.yml", "log_config.yaml"])
 def test_log_config_yaml(
-    mocked_logging_config_module,
-    logging_config,
-    yaml_logging_config,
-    mocker,
-    config_filename,
-):
+    mocked_logging_config_module: MagicMock,
+    logging_config: dict,
+    yaml_logging_config: str,
+    mocker: MockerFixture,
+    config_filename: str,
+) -> None:
     """
     Test that one can load a yaml config from disk.
     """
@@ -244,7 +274,7 @@ def test_log_config_yaml(
     mocked_logging_config_module.dictConfig.assert_called_once_with(logging_config)
 
 
-def test_log_config_file(mocked_logging_config_module):
+def test_log_config_file(mocked_logging_config_module: MagicMock) -> None:
     """
     Test that one can load a configparser config from disk.
     """
@@ -257,20 +287,25 @@ def test_log_config_file(mocked_logging_config_module):
 
 
 @pytest.fixture(params=[0, 1])
-def web_concurrency(request):
-    yield request.param
+def web_concurrency(request: pytest.FixtureRequest) -> typing.Iterator[int]:
+    yield getattr(request, "param")
     if os.getenv("WEB_CONCURRENCY"):
         del os.environ["WEB_CONCURRENCY"]
 
 
 @pytest.fixture(params=["127.0.0.1", "127.0.0.2"])
-def forwarded_allow_ips(request):
-    yield request.param
+def forwarded_allow_ips(request: pytest.FixtureRequest) -> typing.Iterator[str]:
+    yield getattr(request, "param")
     if os.getenv("FORWARDED_ALLOW_IPS"):
         del os.environ["FORWARDED_ALLOW_IPS"]
 
 
-def test_env_file(web_concurrency: int, forwarded_allow_ips: str, caplog, tmp_path):
+def test_env_file(
+    web_concurrency: int,
+    forwarded_allow_ips: str,
+    caplog: pytest.LogCaptureFixture,
+    tmp_path: Path,
+) -> None:
     """
     Test that one can load environment variables using an env file.
     """
@@ -284,7 +319,7 @@ def test_env_file(web_concurrency: int, forwarded_allow_ips: str, caplog, tmp_pa
         config = Config(app=asgi_app, env_file=fp)
         config.load()
 
-    assert config.workers == int(os.getenv("WEB_CONCURRENCY"))
+    assert config.workers == int(str(os.getenv("WEB_CONCURRENCY")))
     assert config.forwarded_allow_ips == os.getenv("FORWARDED_ALLOW_IPS")
     assert len(caplog.records) == 1
     assert f"Loading environment from '{fp}'" in caplog.records[0].message
@@ -297,7 +332,7 @@ def test_env_file(web_concurrency: int, forwarded_allow_ips: str, caplog, tmp_pa
         pytest.param(False, 0, id="access log disabled shouldn't have handlers"),
     ],
 )
-def test_config_access_log(access_log: bool, handlers: int):
+def test_config_access_log(access_log: bool, handlers: int) -> None:
     config = Config(app=asgi_app, access_log=access_log)
     config.load()
 
@@ -306,7 +341,7 @@ def test_config_access_log(access_log: bool, handlers: int):
 
 
 @pytest.mark.parametrize("log_level", [5, 10, 20, 30, 40, 50])
-def test_config_log_level(log_level):
+def test_config_log_level(log_level: int) -> None:
     config = Config(app=asgi_app, log_level=log_level)
     config.load()
 
@@ -316,7 +351,7 @@ def test_config_log_level(log_level):
     assert config.log_level == log_level
 
 
-def test_ws_max_size():
+def test_ws_max_size() -> None:
     config = Config(app=asgi_app, ws_max_size=1000)
     config.load()
     assert config.ws_max_size == 1000
diff --git a/uvicorn/_types.py b/uvicorn/_types.py
@@ -0,0 +1,11 @@
+import types
+import typing
+
+# WSGI
+Environ = typing.MutableMapping[str, typing.Any]
+ExcInfo = typing.Tuple[
+    typing.Type[BaseException], BaseException, typing.Optional[types.TracebackType]
+]
+StartResponse = typing.Callable[
+    [str, typing.Iterable[typing.Tuple[str, str]], typing.Optional[ExcInfo]], None
+]