Subclassing the "runserver" handler doesn't serve static files

[Ivan Voras]

Just for convenience of development, not for production, I'm trying to write a "runserver" lookalike command that also starts Celery, and also uses the autoloader to do it. So I've subclassed Django's runserver Command class and redefined the handler(). It took me ages to understand RUN_MAIN and process management, but here's the result:

import atexit
import errno
import logging
import os
import re
import socket
import subprocess
from time import sleep

from django.conf import settings
from django.core.management.base import CommandError
from django.core.servers.basehttp import run
from django.db import connections
from django.utils import autoreload
from django.utils.regex_helper import _lazy_re_compile
from django.core.management.commands.runserver import Command as RunserverCommand

log = logging.getLogger("glassior")

naiveip_re = _lazy_re_compile(
    r"""^(?:
(?P<addr>
    (?P<ipv4>\d{1,3}(?:\.\d{1,3}){3}) |         # IPv4 address
    (?P<ipv6>\[[a-fA-F0-9:]+\]) |               # IPv6 address
    (?P<fqdn>[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)*) # FQDN
):)?(?P<port>\d+)$""",
    re.X,
)

celery_process = None

class Command(RunserverCommand):
    help = "Starts a lightweight web server for development and a Celery worker with autoreload."

    def handle(self, *args, **options):
        print('runservercelery: Starting both Django and a Celery worker with autoreload...', os.environ.get("RUN_MAIN", False))

        if not settings.DEBUG and not settings.ALLOWED_HOSTS:
            raise CommandError("You must set settings.ALLOWED_HOSTS if DEBUG is False.")

        self.use_ipv6 = options["use_ipv6"]
        if self.use_ipv6 and not socket.has_ipv6:
            raise CommandError("Your Python does not support IPv6.")
        self._raw_ipv6 = False
        if not options["addrport"]:
            self.addr = ""
            self.port = self.default_port
        else:
            m = re.match(naiveip_re, options["addrport"])
            if m is None:
                raise CommandError(
                    '"%s" is not a valid port number '
                    "or address:port pair." % options["addrport"]
                )
            self.addr, _ipv4, _ipv6, _fqdn, self.port = m.groups()
            if not self.port.isdigit():
                raise CommandError("%r is not a valid port number." % self.port)
            if self.addr:
                if _ipv6:
                    self.addr = self.addr[1:-1]
                    self.use_ipv6 = True
                    self._raw_ipv6 = True
                elif self.use_ipv6 and not _fqdn:
                    raise CommandError('"%s" is not a valid IPv6 address.' % self.addr)
        if not self.addr:
            self.addr = self.default_addr_ipv6 if self.use_ipv6 else self.default_addr
            self._raw_ipv6 = self.use_ipv6

        if options["use_reloader"]:
            autoreload.run_with_reloader(self.main_loop, *args, **options)
        else:
            self.main_loop(*args, **options)

    def start_celery(self):
        global celery_process
        if os.environ.get("RUN_MAIN", None) != 'true':
            return

        celery_process = subprocess.Popen(
            'celery -A glassiorapp worker -l info --without-gossip --without-mingle --without-heartbeat -c 1',
            shell=True,
            process_group=0,
        )
        log.info(f"Started celery worker (PID: {celery_process.pid})")


    def our_inner_run(self, *args, **options) -> int | None:
        """
        Taken from django.core.management.commands.runserver.Command.inner_run.
        Returns exit code (None = no error) instead of calling sys.exit().
        """
        # 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 = False # options["use_threading"]
        # 'shutdown_message' is a stealth option.
        shutdown_message = options.get("shutdown_message", "")

        if not options["skip_checks"]:
            self.stdout.write("Performing system checks...\n\n")
            self.check(display_num_errors=True)
        # Need to check migrations here, so can't use the
        # requires_migrations_check attribute.
        self.check_migrations()
        # Close all connections opened during migration checking.
        for conn in connections.all(initialized_only=True):
            conn.close()

        try:
            handler = self.get_handler(*args, **options)
            run(
                self.addr,
                int(self.port),
                handler,
                ipv6=self.use_ipv6,
                threading=threading,
                on_bind=self.on_bind,
                server_cls=self.server_cls,
            )
        except OSError 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 = e
            self.stderr.write("Error: %s" % error_text)
            # Need to use an OS exit because sys.exit doesn't work in a thread
            return 1
        except KeyboardInterrupt:
            print("**** KeyboardInterrupt") # This is never reached.
            if shutdown_message:
                self.stdout.write(shutdown_message)
            return 0

    def main_loop(self, *args, **options):
        self.start_celery()
        exit_code = self.our_inner_run(*args, **options) # Django's code
        # So, apparently our_inner_run doesn't return. 
        print(f"***** our_inner_run exit_code={exit_code}")


@atexit.register
def stop_celery():
    global celery_process
    if celery_process:
        log.info(f"Stopping celery worker (PID: {celery_process.pid})")
        # It's a mess.
        os.system(f"kill -TERM -{celery_process.pid}")
        celery_process = None
        sleep(1)

This works, BUT it doesn't start the static file server. I don't see anything special in the original handler, or in the new one, that would cause this, so I'm just stumped. Switching between running the original runserver and this one, the original serves static files perfectly fine, and this one returns the "no route found" error:

Using the URLconf defined in glassiorapp.urls, Django tried these URL patterns, in this order:

admin/
g/
The current path, static/web/color_modes.js, didn’t match any of these.

You’re seeing this error because you have DEBUG = True in your Django settings file. Change that to False, and Django will display a standard 404 page.

The actual app is being run and responds to the URL endpoints.

Is there something magical about the default runserver?