Bradley Kirton's Blog

#!/usr/bin/env python

import asyncio
import functools
import signal
import logging
import sys


logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger(__name__)


def process_shutdown(signum: int, loop: asyncio.AbstractEventLoop, event: asyncio.Event) -> None:
    """Set the shutdown event.

    :param signum: The signal received.
    :param loop: The running event loop instance.
    :param event: The shutdown event to set.
    """

    logger.info(f"Received {signum} signal, will shutdown soon...")
    loop.remove_signal_handler(signum)
    event.set()


async def amain() -> None:
    """Run the django runserver and tailwind watch processes."""

    loop = asyncio.get_running_loop()
    shutdown_event = asyncio.Event()

    logger.info("Setting up signal handlers...")

    for signum in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(
            signum,
            functools.partial(process_shutdown, signum=signum, loop=loop, event=shutdown_event),
        )

    logger.info("Spawning subprocesses...")

    processes = await asyncio.gather(
        asyncio.subprocess.create_subprocess_shell("python src/server/manage.py runserver localhost:8000"),
        asyncio.subprocess.create_subprocess_shell("tailwindcss -w -i src/server/static/css/_tailwind.css -o src/server/static/css/tailwind.css"),
    )

    await shutdown_event.wait()

    logger.info("Sending SIGTERM to processes...")

    for process in processes:
        process.send_signal(signal.SIGTERM)

    logger.info("Waiting for processes to shutdown...")

    for process in processes:
        await process.wait()


if __name__ == "__main__":
    asyncio.run(amain())