Async support🔗

In this section, we'll look at how you can make you schema support async operations.

Note that asynchronous execution will require an ASGI capable web server.

Setup🔗

To enable async support, you need to set the ASYNC setting to True.

1
2
3
UNDINE = {
    "ASYNC": True,
}

With this, your GraphQL endpoint will change from a sync view to an async view. This allows you to write your Entrypoint resolvers as coroutines.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from undine import Entrypoint, GQLInfo, RootType


class Query(RootType):
    @Entrypoint
    async def example(self, info: GQLInfo) -> str:
        return "foo"

    @example.permissions
    async def permissions(self, info: GQLInfo, value: str) -> None:
        # Some permission check logic here
        return

Various parts of the QueryTypes, MutationTypes, and their Fields and Inputs can also be made async.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from undine import Field, GQLInfo, QueryType

from .models import Task


class TaskType(QueryType[Task]):
    name = Field()

    @name.resolve
    async def resolve_name(self, info: GQLInfo) -> str:
        return self.name

    @name.permissions
    async def permissions(self, info: GQLInfo, value: str) -> None:
        return

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from typing import Any

from undine import GQLInfo, Input, MutationType

from .models import Task


class CustomTaskMutation(MutationType[Task]):
    name = Input()

    @name.validate
    async def validate(self, info: GQLInfo, value: str) -> None:
        return

    @name.permissions
    async def permissions(self, info: GQLInfo, value: str) -> None:
        return

    @classmethod
    async def __mutate__(cls, root: Task, info: GQLInfo, input_data: dict[str, Any]) -> Any:
        return

    @classmethod
    async def __bulk_mutate__(cls, instances: list[Task], info: GQLInfo, input_data: list[dict[str, Any]]) -> Any:
        return

    @classmethod
    async def __permissions__(cls, instance: Task, info: GQLInfo, input_data: dict[str, Any]) -> None:
        return

    @classmethod
    async def __validate__(cls, instance: Task, info: GQLInfo, input_data: dict[str, Any]) -> None:
        return

    @classmethod
    async def __after__(cls, instance: Task, info: GQLInfo, input_data: dict[str, Any]) -> None:
        return

Notes🔗

Using async resolvers without ASYNC enabled will raise an error when an operation resolves using that resolver. Existing resolvers e.g. for QueryTypes and MutationTypes will automatically adapt to work in an async context based on the ASYNC setting.

Another small detail that is worth noting when ASYNC is enabled is that info.context.user is always fetched eagerly, even if it's not used in the operation. This allows using the request user in synchronous parts of the code, like in permission checks (which cannot be made async due to internal implementation details), without causing an error due to using the Django ORM directly in an async context.

Asynchronous execution is also slightly slower than synchronous execution due to inherent overhead of the asyncio event loop.

See Django's async documentation for changes that need to be made for Django to work in async context.