-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathdependencies.py
62 lines (45 loc) · 1.67 KB
/
dependencies.py
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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
__all__ = [
"CURRENT_USER_ID_DEPENDENCY",
"VERIFY_PARSER_DEPENDENCY",
"Shared",
]
from typing import Annotated, Callable, TypeVar, Union, ClassVar, Hashable
from fastapi import Depends
T = TypeVar("T")
CallableOrValue = Union[Callable[[], T], T]
class Shared:
"""
Key-value storage with generic type support for accessing shared dependencies
"""
__slots__ = ()
providers: ClassVar[dict[type, CallableOrValue]] = {}
@classmethod
def register_provider(cls, key: type[T] | Hashable, provider: CallableOrValue):
cls.providers[key] = provider
@classmethod
def f(cls, key: type[T] | Hashable) -> T:
"""
Get shared dependency by key (f - fetch)
"""
if key not in cls.providers:
if isinstance(key, type):
# try by classname
key = key.__name__
if key not in cls.providers:
raise KeyError(f"Provider for {key} is not registered")
elif isinstance(key, str):
# try by classname
for cls_key in cls.providers.keys():
if cls_key.__name__ == key:
key = cls_key
break
else:
raise KeyError(f"Provider for {key} is not registered")
provider = cls.providers[key]
if callable(provider):
return provider()
else:
return provider
from src.api.auth.dependencies import get_current_user_id, verify_parser # noqa: E402
CURRENT_USER_ID_DEPENDENCY = Annotated[str, Depends(get_current_user_id)]
VERIFY_PARSER_DEPENDENCY = Annotated[bool, Depends(verify_parser)]