Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/utils/asyncio.py: 84%

1import asyncio 

2import functools 

3import os 

4 

5from django.core.exceptions import SynchronousOnlyOperation 

6 

7 

8def async_unsafe(message): 

9 """ 

10 Decorator to mark functions as async-unsafe. Someone trying to access 

11 the function while in an async context will get an error message. 

12 """ 

13 

14 def decorator(func): 

15 @functools.wraps(func) 

16 def inner(*args, **kwargs): 

17 if not os.environ.get("DJANGO_ALLOW_ASYNC_UNSAFE"): 17 ↛ 26line 17 didn't jump to line 26, because the condition on line 17 was never false

18 # Detect a running event loop in this thread. 

19 try: 

20 asyncio.get_running_loop() 

21 except RuntimeError: 

22 pass 

23 else: 

24 raise SynchronousOnlyOperation(message) 

25 # Pass onward. 

26 return func(*args, **kwargs) 

27 

28 return inner 

29 

30 # If the message is actually a function, then be a no-arguments decorator. 

31 if callable(message): 31 ↛ 39line 31 didn't jump to line 39, because the condition on line 31 was never false

32 func = message 

33 message = ( 

34 "You cannot call this from an async context - use a thread or " 

35 "sync_to_async." 

36 ) 

37 return decorator(func) 

38 else: 

39 return decorator