Skip to content

runner

dev_tool.services.django.runner

log = logging.getLogger(__name__) module-attribute

DjangoServerRunner

A class for running a Django development server.

This class handles starting, monitoring, and stopping a Django server process.

The constructor for the DjangoServerRunner class.

Parameters:

  • socket_address (str) –

    The socket address for the Django server to bind to.

Source code in dev_tool/services/django/runner.py 
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, socket_address: str) -> None:
    """
    The constructor for the DjangoServerRunner class.

    :param socket_address: The socket address for the Django server to bind to.
    """

    self.socket_address = socket_address
    self.process: subprocess.Popen | None = None
    self.output_thread: threading.Thread | None = None
    self.stop_event = threading.Event()

socket_address = socket_address instance-attribute

process = None instance-attribute

output_thread = None instance-attribute

stop_event = threading.Event() instance-attribute

run

A method that starts and monitors the Django server.

Source code in dev_tool/services/django/runner.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def run(self) -> None:
    """A method that starts and monitors the Django server."""

    creationflags = 0

    if sys.platform == OperatingSystem.WINDOWS:
        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP

    command = [
        VENV_PYTHON,
        BASE / 'manage.py',
        'runserver',
        self.socket_address
    ]

    self.process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        creationflags=creationflags
    )

    self.output_thread = threading.Thread(target=self._stream_output, daemon=True)
    self.output_thread.start()

    handler = signal.getsignal(signal.SIGINT)

    try:
        signal.signal(signal.SIGINT, self._signal_handler)
        signal_registered = True
    except ValueError:
        signal_registered = False

    try:
        while not self.stop_event.is_set():
            if self.process.poll() is not None:
                break

            self.stop_event.wait(timeout=0.1)
    except KeyboardInterrupt:
        self.stop_event.set()
    finally:
        self.stop()
        if signal_registered:
            signal.signal(signal.SIGINT, handler)

stop

A method that stops the Django server and cleans up resources.

Source code in dev_tool/services/django/runner.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def stop(self) -> None:
    """A method that stops the Django server and cleans up resources."""

    self.stop_event.set()

    if self.process is None:
        return

    if self.process.poll() is None:
        self.process.terminate()

        try:
            self.process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.process.kill()

    if self.process.stdout:
        self.process.stdout.close()

    if self.output_thread and not self.output_thread.daemon:
        self.output_thread.join(timeout=1)