django runserver with asyncio

devserver.py

"""
replacement runserver for django with asyncio support
"""
import errno
import os
import sys
import socket
import asyncio
import aiomonitor

from datetime import datetime

from aiohttp import web
from aiohttp_wsgi import WSGIHandler

from django.conf import settings
from django.contrib.staticfiles.handlers import StaticFilesHandler
from django.core.management.commands.runserver import Command as RunserverCommand
from django.utils import autoreload


class Command(RunserverCommand):
    help = "Starts a lightweight Web server for development and also serves static files."

    def add_arguments(self, parser):
        super().add_arguments(parser)
        parser.add_argument(
            '--nostatic',
            action="store_false",
            dest='use_static_handler',
            help='Tells Django to NOT automatically serve static files at STATIC_URL.',
        )
        parser.add_argument(
            '--insecure',
            action="store_true",
            dest='insecure_serving',
            help='Allows serving static files even if DEBUG is False.',
        )

    def get_handler(self, *args, **options):
        handler = super().get_handler(*args, **options)
        use_static_handler = options['use_static_handler']
        insecure_serving = options['insecure_serving']
        if use_static_handler and (settings.DEBUG or insecure_serving):
            return StaticFilesHandler(handler)
        return handler

    def inner_run(self, *args, **options):
        autoreload.raise_last_exception()

        # 'shutdown_message' is a stealth option.
        shutdown_message = options.get('shutdown_message', '')

        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()
        now = datetime.now().strftime('%B %d, %Y - %X')
        self.stdout.write(now)
        addr = self.addr if self._raw_ipv6 else self.addr
        self.stdout.write(
            ("Django version %(version)s, using settings %(settings)r\n"
             "Connect to aiomonitor shell by running `nc %(aiomon_host)s %(aiomon_port)d`\n") % {
                 "aiomon_host": "localhost",
                 "aiomon_port": 50101,
                 "version": self.get_version(),
                 "settings": settings.SETTINGS_MODULE,
             })

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        try:
            # TODO handle exceptions from `django-main-thread`
            with aiomonitor.start_monitor(loop=loop):
                wsgi_handler = WSGIHandler(self.get_handler(*args, **options), url_scheme=self.protocol)
                app = web.Application(loop=loop, debug=settings.DEBUG)
                app.router.add_route("*", "/{path_info:.*}", wsgi_handler)
                web.run_app(app, access_log_format='%a "%r" %s %b %Tf', handle_signals=False, host=addr, port=self.port)
        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 = 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)