Coverage for nexios\dependencies.py: 89%

1from typing import Callable, Dict, List, Optional, Any 

2from inspect import signature, Parameter 

3from functools import wraps 

4import inspect 

5 

6 

7class Depend: 

8 def __init__(self, dependency: Optional[Callable[..., Any]] = None): 

9 self.dependency = dependency 

10 

11 def __class_getitem__(cls, item: Any): 

12 return cls 

13 

14 

15def inject_dependencies(handler: Callable[..., Any]) -> Callable[..., Any]: 

16 """Decorator to inject dependencies into a route handler while preserving parameter names.""" 

17 

18 @wraps(handler) 

19 async def wrapped(*args: List[Any], **kwargs: Dict[str, Any]) -> Any: 

20 sig = signature(handler) 

21 bound_args = sig.bind_partial(*args, **kwargs) 

22 

23 # Get the parameters in order 

24 params = list(sig.parameters.values()) 

25 

26 for param in params: 

27 if ( 

28 param.default != Parameter.empty 

29 and isinstance(param.default, Depend) 

30 and param.name not in bound_args.arguments 

31 ): 

32 depend = param.default 

33 dependency_func = depend.dependency 

34 

35 if dependency_func is None: 

36 raise ValueError( 

37 f"Dependency for parameter '{param.name}' has no provider" 

38 ) 

39 

40 if hasattr(dependency_func, "__wrapped__"): 

41 dependency_func = dependency_func.__wrapped__ # type: ignore[attr-defined] 

42 

43 dep_sig = signature(dependency_func) 

44 dep_kwargs = {} 

45 

46 for dep_param in dep_sig.parameters.values(): 

47 if dep_param.name in bound_args.arguments: 

48 dep_kwargs[dep_param.name] = bound_args.arguments[ 

49 dep_param.name 

50 ] 

51 elif dep_param.default != Parameter.empty and isinstance( 

52 dep_param.default, Depend 

53 ): 

54 nested_dep = dep_param.default.dependency 

55 if inspect.iscoroutinefunction(nested_dep): 

56 dep_kwargs[dep_param.name] = await nested_dep() 

57 else: 

58 dep_kwargs[dep_param.name] = nested_dep() # type: ignore[attr-defined] 

59 

60 # Call the dependency 

61 if inspect.iscoroutinefunction(dependency_func): 

62 bound_args.arguments[param.name] = await dependency_func( 

63 **dep_kwargs 

64 ) 

65 else: 

66 bound_args.arguments[param.name] = dependency_func(**dep_kwargs) 

67 

68 return await handler(**bound_args.arguments) 

69 

70 return wrapped