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

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_configuration_locked", 

18 "registered_session_dependencies", 

19 "test_async_make_session", 

20 "test_make_session", 

21 "database_url", 

22 "make_session", 

23 "session_generator", 

24 "sync_session_generator", 

25 "warn_on_misuse", 

26 "warn_on_uncommitted", 

27 ) 

28 

29 async_database_url: str | None 

30 async_make_session: async_sessionmaker[Any] | None 

31 database_configuration_locked: bool 

32 registered_session_dependencies: set[str] 

33 #: Set by the test fixtures for the duration of a test. The app's dependency 

34 #: overrides and open_session() use these factories before application sources. 

35 test_make_session: sessionmaker[Any] | None 

36 test_async_make_session: async_sessionmaker[Any] | None 

37 database_url: str | None 

38 make_session: sessionmaker[Any] | None 

39 session_generator: Callable[..., AsyncIterator[SA_AsyncSession]] | None 

40 sync_session_generator: Callable[..., Iterator[SA_Session]] | None 

41 warn_on_misuse: bool 

42 warn_on_uncommitted: bool 

43 

44 def __init__(self) -> None: 

45 self.async_database_url = None 

46 self.async_make_session = None 

47 self.database_configuration_locked = False 

48 self.registered_session_dependencies = set() 

49 self.test_make_session = None 

50 self.test_async_make_session = None 

51 self.database_url = None 

52 self.make_session = None 

53 self.session_generator = None 

54 self.sync_session_generator = None 

55 self.warn_on_misuse = False 

56 self.warn_on_uncommitted = True 

57 

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

59 token = _restly_context_ctx.set(self) 

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

61 return self 

62 

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

64 token_stack = _restly_context_token_stack.get() 

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

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

67 token = token_stack[-1] 

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

69 _restly_context_ctx.reset(token) 

70 

71 

72_default_context = RestlyContext() 

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

74 "fastapi_restly_context", default=None 

75) 

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

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

78) 

79 

80 

81def _get_restly_context() -> RestlyContext: 

82 return _restly_context_ctx.get() or _default_context 

83 

84 

85class _FRGlobalsProxy: 

86 def __getattr__(self, name: str): 

87 return getattr(_get_restly_context(), name) 

88 

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

90 setattr(_get_restly_context(), name, value) 

91 

92 

93_fr_globals = _FRGlobalsProxy()