call Twisted from django

main.py

from __future__ import annotations

# with thanks to https://adamj.eu/tech/2020/10/15/a-single-file-rest-api-in-django/

import os
import sys
from dataclasses import dataclass
import treq
import asyncio
import asgiref.sync
import contextlib

from twisted.internet import asyncioreactor, defer, task, main
import hypercorn.config
import hypercorn.asyncio

from django.conf import settings
from django.core.asgi import get_asgi_application
from django.http import HttpResponseRedirect, JsonResponse
from django.urls import path
from django.utils.crypto import get_random_string

settings.configure(
    DEBUG=(os.environ.get("DEBUG", "") == "1"),
    ALLOWED_HOSTS=["*"],  # Disable host header validation
    ROOT_URLCONF=__name__,  # Make this module the urlconf
    SECRET_KEY=get_random_string(
        50
    ),  # We aren't using any security features but Django requires this setting
    MIDDLEWARE=["django.middleware.common.CommonMiddleware"],
)


class DoNotStopAsyncioSelectorReactor(asyncioreactor.AsyncioSelectorReactor):
    def crash(self, *args, **kwargs):
        """
        stop the reactor without stopping the asyncio eventloop
        """
        return super(asyncioreactor.AsyncioSelectorReactor, self).crash(*args, **kwargs)


@contextlib.asynccontextmanager
async def twisted_asyncio_reactor():

    if "twisted.internet.reactor" in sys.modules:
        raise Exception("Twisted reactor already installed")

    reactor = DoNotStopAsyncioSelectorReactor(asyncio.get_running_loop())
    main.installReactor(reactor)

    started = asyncio.Event()
    stopped = asyncio.Event()

    reactor.callWhenRunning(started.set)
    reactor.addSystemEventTrigger("after", "shutdown", stopped.set)
    reactor.startRunning(installSignalHandlers=False)
    await started.wait()
    try:
        yield
    finally:
        reactor.stop()
        await stopped.wait()
        del sys.modules["twisted.internet.reactor"]


async def to_asyncio(corofn, *args, **kwargs):
    return await defer.Deferred.fromCoroutine(corofn(*args, **kwargs)).asFuture(
        asyncio.get_running_loop()
    )


async def treq_get(*args, **kwargs):
    r = await treq.get(*args, **kwargs)
    return await r.json()


async def index(request):
    v = await to_asyncio(treq_get, "https://httpbin.org/json")
    return JsonResponse({"data": v})


urlpatterns = [
    path("", index),
]

app = get_asgi_application()


async def amain():
    async with twisted_asyncio_reactor():
        return await hypercorn.asyncio.serve(app, hypercorn.config.Config())


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