| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061 |
- import asyncio
- import inspect
- import sys
- def is_double_callable(application):
- """
- Tests to see if an application is a legacy-style (double-callable) application.
- """
- # Look for a hint on the object first
- if getattr(application, "_asgi_single_callable", False):
- return False
- if getattr(application, "_asgi_double_callable", False):
- return True
- # Uninstanted classes are double-callable
- if inspect.isclass(application):
- return True
- # Instanted classes depend on their __call__
- if hasattr(application, "__call__"):
- # We only check to see if its __call__ is a coroutine function -
- # if it's not, it still might be a coroutine function itself.
- if asyncio.iscoroutinefunction(application.__call__):
- return False
- # Non-classes we just check directly
- return not asyncio.iscoroutinefunction(application)
- def double_to_single_callable(application):
- """
- Transforms a double-callable ASGI application into a single-callable one.
- """
- async def new_application(scope, receive, send):
- instance = application(scope)
- return await instance(receive, send)
- return new_application
- def guarantee_single_callable(application):
- """
- Takes either a single- or double-callable application and always returns it
- in single-callable style. Use this to add backwards compatibility for ASGI
- 2.0 applications to your server/test harness/etc.
- """
- if is_double_callable(application):
- application = double_to_single_callable(application)
- return application
- if sys.version_info >= (3, 7):
- # these were introduced in 3.7
- get_running_loop = asyncio.get_running_loop
- run_future = asyncio.run
- create_task = asyncio.create_task
- else:
- # marked as deprecated in 3.10, did not exist before 3.7
- get_running_loop = asyncio.get_event_loop
- run_future = asyncio.ensure_future
- # does nothing, this is fine for <3.7
- create_task = lambda task: task
|
