Skip to content

Commit

Permalink
🚨 Cover supervisors/multiprocess.py on mypy (#1059)
Browse files Browse the repository at this point in the history
  • Loading branch information
Kludex authored Jun 1, 2021
1 parent 22f78a7 commit 81c626a
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 7 deletions.
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

0 comments on commit 81c626a

Please sign in to comment.