Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add type annotations to asgiref functions #429

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
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
31 changes: 23 additions & 8 deletions asgiref/compatibility.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
import inspect
from typing import Any, Callable, Coroutine, Optional

from .sync import iscoroutinefunction


def is_double_callable(application):
def is_double_callable(
application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]]
) -> bool:
"""
Tests to see if an application is a legacy-style (double-callable) application.
"""
Expand All @@ -12,10 +15,10 @@ def is_double_callable(application):
return False
if getattr(application, "_asgi_double_callable", False):
return True
# Uninstanted classes are double-callable
# Uninstantiated classes are double-callable
if inspect.isclass(application):
return True
# Instanted classes depend on their __call__
# Instantiated 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.
Expand All @@ -25,22 +28,34 @@ def is_double_callable(application):
return not iscoroutinefunction(application)


def double_to_single_callable(application):
def double_to_single_callable(
application: Callable[..., Callable[[], Coroutine[Any, Any, None]]]
) -> Callable[
[dict[str, Any], Callable[[], Any], Callable[[Any], None]],
Coroutine[Any, Any, None],
]:
"""
Transforms a double-callable ASGI application into a single-callable one.
"""

async def new_application(scope, receive, send):
async def new_application(
scope: dict[str, Any], receive: Callable[[], Any], send: Callable[[Any], None]
) -> None:
instance = application(scope)
return await instance(receive, send)
await instance(receive, send)

return new_application


def guarantee_single_callable(application):
def guarantee_single_callable(
application: Callable[..., Optional[Callable[[], Coroutine[Any, Any, None]]]]
) -> Callable[
[dict[str, Any], Callable[[], Any], Callable[[Any], None]],
Coroutine[Any, Any, None],
]:
"""
Takes either a single- or double-callable application and always returns it
in single-callable style. Use this to add backwards compatibility for ASGI
in single-callable style. Use this to add backward compatibility for ASGI
2.0 applications to your server/test harness/etc.
"""
if is_double_callable(application):
Expand Down
Loading