Coverage for fast_depends/dependencies/provider.py: 100%

1from collections.abc import Callable, Hashable, Iterator 20 ctx1abcdefghijklnopqrsmt

2from contextlib import contextmanager 20 ctx1abcdefghijklnopqrsmt

3from typing import TYPE_CHECKING, Any, TypeAlias 20 ctx1abcdefghijklnopqrsmt

4 

5from fast_depends.core import build_call_model 20 ctx1abcdefghijklnopqrsmt

6 

7if TYPE_CHECKING: 20 ctx1abcdefghijklnopqrsmt

8 from fast_depends.core import CallModel 

9 

10 

11Key: TypeAlias = Hashable 20 ctx1abcdefghijklnopqrsmt

12 

13 

14class Provider: 20 ctx1abcdefghijklnopqrsmt

15 dependencies: dict[Key, "CallModel"] 13 ctx1abcdefghijklm

16 overrides: dict[Key, "CallModel"] 13 ctx1abcdefghijklm

17 

18 def __init__(self) -> None: 20 ctx1abcdefghijklnopqrsmt

19 self.dependencies = {} 20 ctx1abcdefghijklnopqrsmt

20 self.overrides = {} 20 ctx1abcdefghijklnopqrsmt

21 

22 def merge(self, provider: "Provider") -> "Provider": 20 ctx1abcdefghijklnopqrsmt

23 p = Provider() 20 ctx1abcdefghijklnopqrsmt

24 p.dependencies = self.dependencies | provider.dependencies 20 ctx1abcdefghijklnopqrsmt

25 p.overrides = self.overrides | provider.overrides 20 ctx1abcdefghijklnopqrsmt

26 return p 20 ctx1abcdefghijklnopqrsmt

27 

28 def clear(self) -> None: 20 ctx1abcdefghijklnopqrsmt

29 self.overrides = {} 20 ctx1abcdefghijklnopqrsmt

30 

31 def add_dependant( 20 ctx1abcdefghijklnopqrsmt

32 self, 

33 dependant: "CallModel", 

34 ) -> Key: 

35 key = self.__get_original_key(dependant.call) 20 ctx1abcdefghijklnopqrsmt

36 self.dependencies[key] = dependant 20 ctx1abcdefghijklnopqrsmt

37 return key 20 ctx1abcdefghijklnopqrsmt

38 

39 def get_dependant(self, key: Key) -> "CallModel": 20 ctx1abcdefghijklnopqrsmt

40 return self.overrides.get(key) or self.dependencies[key] 20 ctx1abcdefghijklnopqrsmt

41 

42 def override( 20 ctx1abcdefghijklnopqrsmt

43 self, 

44 original: Callable[..., Any], 

45 override: Callable[..., Any], 

46 ) -> None: 

47 key = self.__get_original_key(original) 20 ctx1abcdefghijklnopqrsmt

48 

49 serializer_cls = None 20 ctx1abcdefghijklnopqrsmt

50 

51 if original_dependant := self.dependencies.get(key): 20 ctx1abcdefghijklnopqrsmt

52 serializer_cls = original_dependant.serializer_cls 20 ctx1abcdefghijklnopqrsmt

53 

54 else: 

55 self.dependencies[key] = build_call_model( 20 ctx1abcdefghijklnopqrsmt

56 original, 

57 dependency_provider=self, 

58 ) 

59 

60 override_model = build_call_model( 20 ctx1abcdefghijklnopqrsmt

61 override, 

62 dependency_provider=self, 

63 serializer_cls=serializer_cls, 

64 ) 

65 

66 self.overrides[key] = override_model 20 ctx1abcdefghijklnopqrsmt

67 

68 def __setitem__( 20 ctx1abcdefghijklnopqrsmt

69 self, 

70 key: Callable[..., Any], 

71 value: Callable[..., Any], 

72 ) -> None: 

73 """Alias for `provider[key] = value` syntax""" 

74 self.override(key, value) 20 ctx1abcdefghijklnopqrsmt

75 

76 @contextmanager 20 ctx1abcdefghijklnopqrsmt

77 def scope( 20 ctx1abcdefghijklnopqrsmt

78 self, 

79 original: Callable[..., Any], 

80 override: Callable[..., Any], 

81 ) -> Iterator[None]: 

82 self.override(original, override) 20 ctx1abcdefghijklnopqrsmt

83 try: 20 ctx1abcdefghijklnopqrsmt

84 yield 20 ctx1abcdefghijklnopqrsmt

85 finally: 

86 self.overrides.pop(self.__get_original_key(original), None) 20 ctx1abcdefghijklnopqrsmt

87 

88 def __get_original_key(self, original: Callable[..., Any]) -> Key: 20 ctx1abcdefghijklnopqrsmt

89 return original 20 ctx1abcdefghijklnopqrsmt