Coverage for fastapi_restly / db / _globals.py: 96%

1from collections.abc import AsyncIterator, Callable, Iterator 

2from contextvars import ContextVar, Token 

3from typing import Any 

4 

5from sqlalchemy.ext.asyncio import AsyncSession as SA_AsyncSession 

6from sqlalchemy.ext.asyncio import async_sessionmaker 

7from sqlalchemy.orm import Session as SA_Session 

8from sqlalchemy.orm import sessionmaker 

9 

10 

11class RestlyContext: 

12 """Private container for Restly runtime state used by tests and internals.""" 

13 

14 __slots__ = ( 

15 "async_database_url", 

16 "async_make_session", 

17 "database_url", 

18 "make_session", 

19 "session_generator", 

20 "sync_session_generator", 

21 "warn_on_misuse", 

22 "warn_on_uncommitted", 

23 ) 

24 

25 async_database_url: str | None 

26 async_make_session: async_sessionmaker[Any] | None 

27 database_url: str | None 

28 make_session: sessionmaker[Any] | None 

29 session_generator: Callable[[], AsyncIterator[SA_AsyncSession]] | None 

30 sync_session_generator: Callable[[], Iterator[SA_Session]] | None 

31 warn_on_misuse: bool 

32 warn_on_uncommitted: bool 

33 

34 def __init__(self) -> None: 

35 self.async_database_url = None 

36 self.async_make_session = None 

37 self.database_url = None 

38 self.make_session = None 

39 self.session_generator = None 

40 self.sync_session_generator = None 

41 self.warn_on_misuse = False 

42 self.warn_on_uncommitted = True 

43 

44 def __enter__(self) -> "RestlyContext": 

45 token = _restly_context_ctx.set(self) 

46 _restly_context_token_stack.set(_restly_context_token_stack.get() + (token,)) 

47 return self 

48 

49 def __exit__(self, *exc_info: object) -> None: 

50 token_stack = _restly_context_token_stack.get() 

51 if not token_stack: 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true

52 raise RuntimeError("RestlyContext was exited without being entered.") 

53 token = token_stack[-1] 

54 _restly_context_token_stack.set(token_stack[:-1]) 

55 _restly_context_ctx.reset(token) 

56 

57 

58_default_context = RestlyContext() 

59_restly_context_ctx: ContextVar[RestlyContext | None] = ContextVar( 

60 "fastapi_restly_context", default=None 

61) 

62_restly_context_token_stack: ContextVar[tuple[Token[RestlyContext | None], ...]] = ( 

63 ContextVar("fastapi_restly_context_token_stack", default=()) 

64) 

65 

66 

67def _get_restly_context() -> RestlyContext: 

68 return _restly_context_ctx.get() or _default_context 

69 

70 

71class _FRGlobalsProxy: 

72 def __getattr__(self, name: str): 

73 return getattr(_get_restly_context(), name) 

74 

75 def __setattr__(self, name: str, value): 

76 setattr(_get_restly_context(), name, value) 

77 

78 

79_fr_globals = _FRGlobalsProxy()