Skip to content

Instantly share code, notes, and snippets.

@graingert
Created May 11, 2021
Embed
What would you like to do?
call Twisted from django
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())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment