diff --git a/setup.cfg b/setup.cfg index aabc3dbf1..3b6b48874 100644 --- a/setup.cfg +++ b/setup.cfg @@ -12,6 +12,12 @@ files = uvicorn/supervisors/basereload.py, uvicorn/importer.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, uvicorn/middleware/debug.py, uvicorn/supervisors/watchgodreload.py, uvicorn/logging.py, diff --git a/uvicorn/loops/asyncio.py b/uvicorn/loops/asyncio.py index 223e76dee..ba764b03a 100644 --- a/uvicorn/loops/asyncio.py +++ b/uvicorn/loops/asyncio.py @@ -4,7 +4,8 @@ import sys -def asyncio_setup(): # pragma: no cover +def asyncio_setup() -> None: # pragma: no cover + loop: asyncio.AbstractEventLoop if ( sys.version_info.major >= 3 and sys.version_info.minor >= 8 diff --git a/uvicorn/loops/auto.py b/uvicorn/loops/auto.py index d550c3290..d587b6229 100644 --- a/uvicorn/loops/auto.py +++ b/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 diff --git a/uvicorn/loops/uvloop.py b/uvicorn/loops/uvloop.py index 8cd55c77d..cb6f32638 100644 --- a/uvicorn/loops/uvloop.py +++ b/uvicorn/loops/uvloop.py @@ -3,5 +3,5 @@ import uvloop -def uvloop_setup(): +def uvloop_setup() -> None: asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) diff --git a/uvicorn/main.py b/uvicorn/main.py index e85558b4d..424b388d8 100644 --- a/uvicorn/main.py +++ b/uvicorn/main.py @@ -5,6 +5,7 @@ import typing import click +from asgiref.typing import ASGIApplication import uvicorn from uvicorn.config import ( @@ -21,17 +22,17 @@ from uvicorn.server import Server, ServerState # noqa: F401 # Used to be defined here. from uvicorn.supervisors import ChangeReload, Multiprocess -LEVEL_CHOICES = click.Choice(LOG_LEVELS.keys()) -HTTP_CHOICES = click.Choice(HTTP_PROTOCOLS.keys()) -WS_CHOICES = click.Choice(WS_PROTOCOLS.keys()) -LIFESPAN_CHOICES = click.Choice(LIFESPAN.keys()) +LEVEL_CHOICES = click.Choice(list(LOG_LEVELS.keys())) +HTTP_CHOICES = click.Choice(list(HTTP_PROTOCOLS.keys())) +WS_CHOICES = click.Choice(list(WS_PROTOCOLS.keys())) +LIFESPAN_CHOICES = click.Choice(list(LIFESPAN.keys())) LOOP_CHOICES = click.Choice([key for key in LOOP_SETUPS.keys() if key != "none"]) INTERFACE_CHOICES = click.Choice(INTERFACES) 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( @@ -288,7 +289,7 @@ def print_version(ctx, param, value): show_default=True, ) def main( - app, + app: str, host: str, port: int, uds: str, @@ -326,11 +327,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, @@ -368,10 +368,10 @@ def main( "use_colors": use_colors, "factory": factory, } - run(**kwargs) + run(app, **kwargs) -def run(app, **kwargs): +def run(app: typing.Union[ASGIApplication, str], **kwargs: typing.Any) -> None: config = Config(app, **kwargs) server = Server(config=config) @@ -385,12 +385,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() diff --git a/uvicorn/protocols/http/auto.py b/uvicorn/protocols/http/auto.py index 752a02ddd..1aa996744 100644 --- a/uvicorn/protocols/http/auto.py +++ b/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 diff --git a/uvicorn/protocols/websockets/auto.py b/uvicorn/protocols/websockets/auto.py index e3364dc73..0dfba3bdb 100644 --- a/uvicorn/protocols/websockets/auto.py +++ b/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 diff --git a/uvicorn/supervisors/__init__.py b/uvicorn/supervisors/__init__.py index 7a2b7b555..5a27e35f4 100644 --- a/uvicorn/supervisors/__init__.py +++ b/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"] diff --git a/uvicorn/workers.py b/uvicorn/workers.py index ddad5db0e..4ffe2ebd9 100644 --- a/uvicorn/workers.py +++ b/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": "auto", "http": "auto"} - 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()