fix: mypy issues

setup.cfg

@@ -10,7 +10,13 @@ files =
     uvicorn/lifespan,
     tests/test_lifespan.py,
     uvicorn/importer.py,
-    uvicorn/protocols/utils.py
+    uvicorn/protocols/utils.py,
+    uvicorn/loops,
+    uvicorn/main.py,
+    uvicorn/workers.py,
+    uvicorn/protocols/http/auto.py,
+    uvicorn/protocols/websockets/auto.py,
+    uvicorn/supervisors/__init__.py
 
 
 [mypy-tests.*]

uvicorn/loops/asyncio.py

@@ -4,7 +4,8 @@
 import sys
 
 
-def asyncio_setup():
+def asyncio_setup() -> None:
+    loop: asyncio.AbstractEventLoop
     if (
         sys.version_info.major >= 3
         and sys.version_info.minor >= 8

uvicorn/loops/auto.py

@@ -1,4 +1,4 @@
-def auto_loop_setup():
+def auto_loop_setup() -> None:
     try:
         import uvloop  # noqa
     except ImportError:  # pragma: no cover

uvicorn/loops/uvloop.py

@@ -3,5 +3,5 @@
 import uvloop
 
 
-def uvloop_setup():
+def uvloop_setup() -> None:
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

uvicorn/main.py

@@ -31,7 +31,7 @@
 logger = logging.getLogger("uvicorn.error")
 
 
-def print_version(ctx, param, value):
+def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> None:
     if not value or ctx.resilient_parsing:
         return
     click.echo(
@@ -281,7 +281,7 @@ def print_version(ctx, param, value):
     show_default=True,
 )
 def main(
-    app,
+    app: str,
     host: str,
     port: int,
     uds: str,
@@ -318,11 +318,10 @@ def main(
     use_colors: bool,
     app_dir: str,
     factory: bool,
-):
+) -> None:
     sys.path.insert(0, app_dir)
 
     kwargs = {
-        "app": app,
         "host": host,
         "port": port,
         "uds": uds,
@@ -359,10 +358,10 @@ def main(
         "use_colors": use_colors,
         "factory": factory,
     }
-    run(**kwargs)
+    run(app, **kwargs)
 
 
-def run(app, **kwargs):
+def run(app: str, **kwargs: typing.Any) -> None:
     config = Config(app, **kwargs)
     server = Server(config=config)
 
@@ -376,12 +375,10 @@ def run(app, **kwargs):
 
     if config.should_reload:
         sock = config.bind_socket()
-        supervisor = ChangeReload(config, target=server.run, sockets=[sock])
-        supervisor.run()
+        ChangeReload(config, target=server.run, sockets=[sock]).run()
     elif config.workers > 1:
         sock = config.bind_socket()
-        supervisor = Multiprocess(config, target=server.run, sockets=[sock])
-        supervisor.run()
+        Multiprocess(config, target=server.run, sockets=[sock]).run()
     else:
         server.run()
 

uvicorn/protocols/http/auto.py

@@ -1,3 +1,7 @@
+import asyncio
+from typing import Type
+
+AutoHTTPProtocol: Type[asyncio.Protocol]
 try:
     import httptools  # noqa
 except ImportError:  # pragma: no cover

uvicorn/protocols/websockets/auto.py

@@ -1,3 +1,7 @@
+import asyncio
+import typing
+
+AutoWebSocketsProtocol: typing.Optional[typing.Type[asyncio.Protocol]]
 try:
     import websockets  # noqa
 except ImportError:  # pragma: no cover

uvicorn/supervisors/__init__.py

@@ -1,8 +1,14 @@
+import typing
+
+from uvicorn.supervisors.basereload import BaseReload
 from uvicorn.supervisors.multiprocess import Multiprocess
 
-try:
-    from uvicorn.supervisors.watchgodreload import WatchGodReload as ChangeReload
-except ImportError:  # pragma: no cover
-    from uvicorn.supervisors.statreload import StatReload as ChangeReload
+if typing.TYPE_CHECKING:
+    ChangeReload: typing.Type[BaseReload]  # pragma: no cover
+else:
+    try:
+        from uvicorn.supervisors.watchgodreload import WatchGodReload as ChangeReload
+    except ImportError:  # pragma: no cover
+        from uvicorn.supervisors.statreload import StatReload as ChangeReload
 
 __all__ = ["Multiprocess", "ChangeReload"]

uvicorn/workers.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import signal
+from typing import Any
 
 from gunicorn.workers.base import Worker
 
@@ -16,7 +17,7 @@ class UvicornWorker(Worker):
 
     CONFIG_KWARGS = {"loop": "uvloop", "http": "httptools"}
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         super(UvicornWorker, self).__init__(*args, **kwargs)
 
         logger = logging.getLogger("uvicorn.error")
@@ -58,24 +59,24 @@ def __init__(self, *args, **kwargs):
 
         self.config = Config(**config_kwargs)
 
-    def init_process(self):
+    def init_process(self) -> None:
         self.config.setup_event_loop()
         super(UvicornWorker, self).init_process()
 
-    def init_signals(self):
+    def init_signals(self) -> None:
         # Reset signals so Gunicorn doesn't swallow subprocess return codes
         # other signals are set up by Server.install_signal_handlers()
         # See: https://github.com/encode/uvicorn/issues/894
         for s in self.SIGNALS:
             signal.signal(s, signal.SIG_DFL)
 
-    def run(self):
+    def run(self) -> None:
         self.config.app = self.wsgi
         server = Server(config=self.config)
         loop = asyncio.get_event_loop()
         loop.run_until_complete(server.serve(sockets=self.sockets))
 
-    async def callback_notify(self):
+    async def callback_notify(self) -> None:
         self.notify()