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
38
39
|
import os
from asyncio import get_running_loop
from functools import wraps
from django.core.exceptions import SynchronousOnlyOperation
def async_unsafe(message):
"""
Decorator to mark functions as async-unsafe. Someone trying to access
the function while in an async context will get an error message.
"""
def decorator(func):
@wraps(func)
def inner(*args, **kwargs):
# Detect a running event loop in this thread.
try:
get_running_loop()
except RuntimeError:
pass
else:
if not os.environ.get("DJANGO_ALLOW_ASYNC_UNSAFE"):
raise SynchronousOnlyOperation(message)
# Pass onward.
return func(*args, **kwargs)
return inner
# If the message is actually a function, then be a no-arguments decorator.
if callable(message):
func = message
message = (
"You cannot call this from an async context - use a thread or "
"sync_to_async."
)
return decorator(func)
else:
return decorator
|
