run.py

# To make this work, you may need to increase value in
#   /proc/sys/fs/inotify/max_user_instances
#
# To persist this value on Debian/Ubuntu, run
#   echo fs.inotify.max_user_instances=8192 | sudo tee -a /etc/sysctl.conf && sudo sysctl -p

import os
import traceback
from threading import Event
from threading import Thread

from django.contrib.staticfiles import finders
from django.contrib.staticfiles.management.commands.runserver import Command as RunServer
from django.core.management import call_command
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer


def collectstatic(clear=False):
    print('Running collectstatic...')
    try:
        call_command('collectstatic', interactive=False, clear=clear, verbosity=0 if clear else 1)
    except Exception as e:
        traceback.print_exc()
        message = e.message if hasattr(e, 'message') else repr(e)
        if hasattr(message, 'decode'):
            message = message.decode()
        print(message)


class CollectStaticThread(Thread):
    def __init__(self, delay):
        super().__init__()
        self.delay = delay
        self.triggered = Event()
        self.timer = Event()
        self.stopped = Event()

    def run(self):
        while not self.stopped.is_set():
            self.triggered.wait()
            self.timer.wait(self.delay)
            if self.triggered.is_set() and not self.stopped.is_set():
                self.triggered.clear()
                collectstatic()

    def trigger(self):
        self.triggered.set()

    def stop(self):
        print('Stopping watcher')
        self.stopped.set()
        self.triggered.set()
        self.timer.set()

    def join(self, timeout=None):
        self.stop()
        super().join(timeout)


class StaticFileSystemEventHandler(FileSystemEventHandler):
    def on_any_event(self, event):
        if hasattr(collectstatic, 'thread'):
            collectstatic.thread.trigger()


class Command(RunServer):
    help = 'Run server and collect static in background when any static file changes'

    def add_arguments(self, parser):
        super().add_arguments(parser)
        parser.add_argument('--delay', action='store', type=int, default=3)
        parser.add_argument('--collectstatic', action='store_true')
        parser.add_argument('--compilemessages', action='store_true')
        parser.add_argument('--clear_cache', action='store_true')

    def handle(self, *args, **options):
        if os.environ.get('RUN_MAIN') != 'true':
            if options['collectstatic']:
                collectstatic(clear=True)
            if options['compilemessages']:
                call_command('compilemessages')
            if options['clear_cache']:
                call_command('clear_cache')
            static_roots = self.find_static_roots()
            if static_roots:
                observer = Observer()
                event_handler = StaticFileSystemEventHandler()
                print('Watching:')
                for path in static_roots:
                    print(path)
                    observer.schedule(event_handler, str(path), recursive=True)
                collectstatic.thread = CollectStaticThread(delay=options['delay'])
                collectstatic.thread.start()
                observer.start()
            else:
                print('No static roots have been found')
        super().handle(*args, **options)

    @staticmethod
    def find_static_roots():
        found_files = set()
        for finder in finders.get_finders():
            for path, storage in finder.list([]):
                found_files.add(storage.base_location)
        return found_files