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

38 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1from enum import StrEnum 1ab

2from inspect import Parameter, Signature 1ab

3from typing import Any 1ab

4 

5from fastapi import Query 1ab

6from makefun import with_signature 1ab

7 

8from polar.exceptions import PolarRequestValidationError 1ab

9 

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

11 

12 

13class _SortingGetter[PE: StrEnum]: 1ab

14 def __init__( 1ab

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

16 ) -> None: 

17 self.sort_property_enum = sort_property_enum 1ab

18 self.default_sorting = default_sorting 1ab

19 

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

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

22 sorting = self.default_sorting 

23 

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

25 for criteria in sorting: 1c

26 desc = False 1c

27 if criteria[0] == "-": 27 ↛ 30line 27 didn't jump to line 30 because the condition on line 27 was always true1c

28 desc = True 1c

29 criteria = criteria[1:] 1c

30 try: 1c

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

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 1c

44 

45 

46def SortingGetter[PE: StrEnum]( 1ab

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 = [] 1ab

58 for value in sort_property_enum: 1ab

59 enum_values.append(value.value) 1ab

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

61 

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

63 sort_property_enum.__name__, 

64 enum_values, 

65 ) 

66 

67 parameters: list[Parameter] = [ 1ab

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) 1ab

84 

85 class _SortingGetterSignature(_SortingGetter[Any]): 1ab

86 @with_signature(signature) 1ab

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

88 return await super().__call__(sorting) 1c

89 

90 return _SortingGetterSignature(sort_property_enum, default_sorting) 1ab