Skip to content

Instantly share code, notes, and snippets.

@aputs
Last active Sep 10, 2019
Embed
What would you like to do?
django runserver with asyncio
"""
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment