zulip/zilencer/management/commands/rundjangoserver.py

import errno
import os
import socket
import sys
from typing import Any, NoReturn

from django.conf import settings
from django.core.management.commands.runserver import Command as DjangoCommand
from django.core.servers.basehttp import run
from django.utils import autoreload


class Command(DjangoCommand):
    # Copied from Django's runserver and modified to remove print statements.
    # This lets us declutter run-dev.py startup output.
    def inner_run(self, *args: Any, **options: Any) -> None:
        # If an exception was silenced in ManagementUtility.execute in order
        # to be raised in the child process, raise it now.
        autoreload.raise_last_exception()

        threading = options["use_threading"]
        # 'shutdown_message' is a stealth option.
        shutdown_message = options.get("shutdown_message", "")
        quit_command = "CTRL-BREAK" if sys.platform == "win32" else "CONTROL-C"

        self.check(display_num_errors=False)
        # Need to check migrations here, so can't use the
        # requires_migrations_check attribute.
        self.check_migrations()
        self.stdout.write(
            ("Django process (re)started. Quit the server with %(quit_command)s.\n")
            % {
                "version": self.get_version(),
                "settings": settings.SETTINGS_MODULE,
                "protocol": self.protocol,
                "addr": "[%s]" % self.addr if self._raw_ipv6 else self.addr,
                "port": self.port,
                "quit_command": quit_command,
            }
        )

        try:
            handler = self.get_handler(*args, **options)
            run(
                self.addr,
                int(self.port),
                handler,
                ipv6=self.use_ipv6,
                threading=threading,
                server_cls=self.server_cls,
            )
        except socket.error as e:
            # Use helpful error messages instead of ugly tracebacks.
            ERRORS = {
                errno.EACCES: "You don't have permission to access that port.",
                errno.EADDRINUSE: "That port is already in use.",
                errno.EADDRNOTAVAIL: "That IP address can't be assigned to.",
            }
            try:
                error_text = ERRORS[e.errno]
            except KeyError:
                error_text = str(e)
            self.stderr.write("Error: %s" % error_text)
            # Need to use an OS exit because sys.exit doesn't work in a thread
            os._exit(1)
        except KeyboardInterrupt:
            if shutdown_message:
                self.stdout.write(shutdown_message)
            sys.exit(0)