Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 21 additions & 6 deletions asgiref/compatibility.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,24 @@
from __future__ import annotations

import inspect
from typing import Any, Awaitable, Callable, Union, cast

from typing_extensions import TypeAlias

from .sync import iscoroutinefunction

Scope: TypeAlias = dict[str, Any]
Receive: TypeAlias = Callable[[], Awaitable[Any]]
Send: TypeAlias = Callable[[dict[str, Any]], Awaitable[None]]

ASGISingleCallable: TypeAlias = Callable[[Scope, Receive, Send], Awaitable[Any]]
ASGIDoubleCallableInstance: TypeAlias = Callable[[Receive, Send], Awaitable[Any]]
ASGIDoubleCallable: TypeAlias = Callable[[Scope], ASGIDoubleCallableInstance]

ASGIApplication: TypeAlias = Union[ASGISingleCallable, ASGIDoubleCallable, type]


def is_double_callable(application):
def is_double_callable(application: ASGIApplication) -> bool:
"""
Tests to see if an application is a legacy-style (double-callable) application.
"""
Expand All @@ -25,24 +40,24 @@ def is_double_callable(application):
return not iscoroutinefunction(application)


def double_to_single_callable(application):
def double_to_single_callable(application: ASGIDoubleCallable) -> ASGISingleCallable:
"""
Transforms a double-callable ASGI application into a single-callable one.
"""

async def new_application(scope, receive, send):
async def new_application(scope: Scope, receive: Receive, send: Send) -> Any:
instance = application(scope)
return await instance(receive, send)

return new_application


def guarantee_single_callable(application):
def guarantee_single_callable(application: ASGIApplication) -> ASGISingleCallable:
"""
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
return double_to_single_callable(cast(ASGIDoubleCallable, application))
return cast(ASGISingleCallable, application)