Coverage for polar/kit/sorting.py: 91%

1from enum import StrEnum 

2from inspect import Parameter, Signature 

3from typing import Any 

4 

5from fastapi import Query 

6from makefun import with_signature 

7 

8from polar.exceptions import PolarRequestValidationError 

9 

10type Sorting[PE] = tuple[PE, bool] 

11 

12 

13class _SortingGetter[PE: StrEnum]: 

14 def __init__( 

15 self, sort_property_enum: type[PE], default_sorting: list[str] 

16 ) -> None: 

17 self.sort_property_enum = sort_property_enum 

18 self.default_sorting = default_sorting 

19 

20 async def __call__(self, sorting: list[str] | None) -> list[Sorting[PE]]: 

21 if sorting is None: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 sorting = self.default_sorting 

23 

24 parsed_sorting: list[tuple[PE, bool]] = [] 

25 for criteria in sorting: 

26 desc = False 

27 if criteria[0] == "-": 

28 desc = True 

29 criteria = criteria[1:] 

30 try: 

31 parsed_sorting.append((self.sort_property_enum(criteria), desc)) 

32 except ValueError as e: 

33 raise PolarRequestValidationError( 

34 [ 

35 { 

36 "loc": ("query", "sorting"), 

37 "input": criteria, 

38 "msg": "Invalid sorting criterion.", 

39 "type": "enum", 

40 } 

41 ] 

42 ) 

43 return parsed_sorting 

44 

45 

46def SortingGetter[PE: StrEnum]( 

47 sort_property_enum: type[PE], default_sorting: list[str] 

48) -> _SortingGetter[PE]: 

49 """ 

50 Here comes some blood magic 🧙‍♂️ 

51 

52 Generate a version of `_SortingGetter` with an overriden `__call__` signature. 

53 

54 By doing so, we can dynamically inject the allowed sorting properties into FastAPI 

55 dependency, so they are properrly detected by the OpenAPI generator. 

56 """ 

57 enum_values = [] 

58 for value in sort_property_enum: 

59 enum_values.append(value.value) 

60 enum_values.append(f"-{value.value}") 

61 

62 sort_property_full_enum = StrEnum( # type: ignore[misc] 

63 sort_property_enum.__name__, 

64 enum_values, 

65 ) 

66 

67 parameters: list[Parameter] = [ 

68 Parameter(name="self", kind=Parameter.POSITIONAL_OR_KEYWORD), 

69 Parameter( 

70 name="sorting", 

71 kind=Parameter.POSITIONAL_OR_KEYWORD, 

72 default=Query( 

73 default_sorting, 

74 description=( 

75 "Sorting criterion. " 

76 "Several criteria can be used simultaneously and will be applied in order. " 

77 "Add a minus sign `-` before the criteria name to sort by descending order." 

78 ), 

79 ), 

80 annotation=list[sort_property_full_enum] | None, 

81 ), 

82 ] 

83 signature = Signature(parameters) 

84 

85 class _SortingGetterSignature(_SortingGetter[Any]): 

86 @with_signature(signature) 

87 async def __call__(self, sorting: Any) -> list[Sorting[Any]]: 

88 return await super().__call__(sorting) 

89 

90 return _SortingGetterSignature(sort_property_enum, default_sorting)