How do I start/stop Hypercorn/Uvicorn server as a background task for an async application (like a discord bot) Python

I am currently creating a django application w/ asgi and I am currently having problems of setting up hypercorn and uvicorn to run in background with graceful shutdown. When I set up my application from asgi to run on hypercorn only using asyncio.create_task and starting it only, the website doesn't run.

Hypercorn code snippet:

from scripts import funcs
import nextcord
from nextcord.ext import commands
from nextcord import Interaction
import asyncio
# from uvicorn import Config, Server
# import uvicorn
import subprocess
from subprocess import CREATE_NEW_CONSOLE
import signal
# import multiprocessing
import nest_asyncio
import os
import sys
sys.path.insert(1, 'C:\\Users\\Sub01\\Project\\PaulWebsite\\app')
from hypercorn.config import Config
from hypercorn.asyncio import serve
from hypercorn.run import run
import hypercorn
import asyncio
from paul_site.asgi import application
import signal

nest_asyncio.apply()
createEmbed = funcs.embedCreator()

shutdown_event = asyncio.Event()

def _signal_handler(*_) -> None:
    shutdown_event.set()

class HYPERCORN:
    config = Config()
    coro = None
    
    def __init__(self) -> None:
        self.config.from_object("paul_site.asgi")
        self.evtLoop = asyncio.new_event_loop()
    
    async def start(self):
        self.coro = self.evtLoop.create_task(await serve(application, self.config))

    def stop(self):
        self.evtLoop.add_signal_handler(signal.SIGINT, _signal_handler)
        self.evtLoop.run_until_complete(
            asyncio.to_thread(serve(application, self.config, shutdown_trigger=shutdown_event.wait))
        )


class baseCommand(commands.Cog):
    proc = None

    def __init__(self, client):
        self.client = client
        self.website = HYPERCORN()

    @nextcord.slash_command()
    async def bot(self, interaction: Interaction):
        pass
    
    @bot.subcommand(description="Stops the bot")
    async def shutdown(self, interaction: Interaction):
        await interaction.response.send_message(embed=createEmbed.createEmbed(title="Exit", description="Bot's down", footer=f"Requested by {interaction.user.name}"))
        exit()

    # Create command group site
    @nextcord.slash_command()
    async def site(self, interaction: Interaction):
        pass
    
    @site.subcommand(description="Starts the website")
    async def start(self, interaction: Interaction):
        try:
            await self.website.start()
            await interaction.response.send_message(embed=createEmbed.createEmbed(title="Start Website", description=f"""
                                                                                  **Website started successfully**
                                                                                  """, footer=f"Requested by {interaction.user.name}"))
        except Exception as e:
            await interaction.response.send_message(
                embed=createEmbed.createEmbed(title="Start Website Error", description=
                f"""
                ```bash
                {e}
                ```
                """, footer=f"Requested by {interaction.user.name}")
            )

    @site.subcommand(description='Stops the website')
    async def stop(self, interaction: Interaction):
        self.website.stop()
        await interaction.followup.send(embed=createEmbed.createEmbed(title="Stop Website", description=f"""
                                                                              **Website stopped successfully!**
                                                                              """, footer=f"Requested by {interaction.user.name}"))
        del self.proc

def setup(client):
    client.add_cog(baseCommand(client))

Uvicorn code snippet:

import sys
sys.path.insert(1, 'C:\\Users\\Sub01\\Project\\PaulWebsite\\app')
import asyncio
from paul_site.asgi import application
import signal
import time
import uvicorn
from multiprocessing import Process

class UvicornServer(uvicorn.Server):
    
    def __init__(self, host: str = "127.0.0.1", port: int = 8000):
        self.host = host
        self.port = port
    
    async def setup(self):
        self.proc = Process(
            target=uvicorn.run,
            args=[application],
            kwargs={
                'host': self.host,
                'port': self.port,
            },
            daemon=True
        )
        # self.proc.run()
        await self.proc.start()
        await asyncio.sleep(0.5)
    
    async def down(self):
        self.proc.terminate()

def blockingFunc():
    prevTime = time.time()
    while True:
        print("Elapsed time: ", time.time() - prevTime)
        time.sleep(1)
        if time.time() - prevTime >= 4:
            break

async def main():
    server = UvicornServer()
    await server.setup()
    blockingFunc()
    await server.down()

asyncio.run(main())

Asgi.py:

"""
ASGI config for paul_site project.

It exposes the ASGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/4.1/howto/deployment/asgi/
"""

import os

from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from paul_site_app.ws_urlpatterns import ws_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'paul_site.settings')

application = ProtocolTypeRouter({
    'http': get_asgi_application(),
    'websocket': URLRouter(ws_urlpatterns)
})

Looking at the examples from people incorporating FastAPI and running uvicorn as a background task, I tried it but it only results in a runtime error. I've also tried having a command open a terminal and running the application via cli but soon realized that the code that invokes a new terminal isn't compatible with different platforms.

Back to Top