Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

🚨 Cover supervisors/multiprocess.py on mypy #1059

Merged
merged 1 commit into from
Jun 1, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion setup.cfg
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@ files =
uvicorn/_handlers,
uvicorn/__init__.py,
uvicorn/__main__.py,
uvicorn/subprocess.py
uvicorn/subprocess.py,
uvicorn/supervisors/multiprocess.py


[mypy-tests.*]
Expand Down
22 changes: 16 additions & 6 deletions uvicorn/supervisors/multiprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,14 @@
import os
import signal
import threading
from multiprocessing.context import SpawnProcess
from socket import socket
from types import FrameType
from typing import Callable, List, Optional

import click

from uvicorn.config import Config
from uvicorn.subprocess import get_subprocess

HANDLED_SIGNALS = (
Expand All @@ -16,26 +21,31 @@


class Multiprocess:
def __init__(self, config, target, sockets):
def __init__(
self,
config: Config,
target: Callable[[Optional[List[socket]]], None],
sockets: List[socket],
) -> None:
self.config = config
self.target = target
self.sockets = sockets
self.processes = []
self.processes: List[SpawnProcess] = []
self.should_exit = threading.Event()
self.pid = os.getpid()

def signal_handler(self, sig, frame):
def signal_handler(self, sig: signal.Signals, frame: FrameType) -> None:
"""
A signal handler that is registered with the parent process.
"""
self.should_exit.set()

def run(self):
def run(self) -> None:
self.startup()
self.should_exit.wait()
self.shutdown()

def startup(self):
def startup(self) -> None:
message = "Started parent process [{}]".format(str(self.pid))
color_message = "Started parent process [{}]".format(
click.style(str(self.pid), fg="cyan", bold=True)
Expand All @@ -52,7 +62,7 @@ def startup(self):
process.start()
self.processes.append(process)

def shutdown(self):
def shutdown(self) -> None:
for process in self.processes:
process.join()

Expand Down