Coverage for dibbler / queries / query_helpers.py: 83%

1from datetime import datetime 

2from typing import TypeVar 

3 

4from sqlalchemy import ( 

5 BindParameter, 

6 ColumnExpressionArgument, 

7 literal, 

8 select, 

9) 

10from sqlalchemy.orm import QueryableAttribute 

11 

12from dibbler.models import Transaction 

13 

14T = TypeVar("T") 

15 

16 

17def const(value: T) -> BindParameter[T]: 

18 """ 

19 Create a constant SQL literal bind parameter. 

20 

21 This is useful to avoid too many `?` bind parameters in SQL queries, 

22 when the input value is known to be safe. 

23 """ 

24 

25 return literal(value, literal_execute=True) 

26 

27 

28CONST_ZERO: BindParameter[int] = const(0) 

29"""A constant SQL expression `0`. This will render as a literal `0` in SQL queries.""" 

30 

31CONST_ONE: BindParameter[int] = const(1) 

32"""A constant SQL expression `1`. This will render as a literal `1` in SQL queries.""" 

33 

34CONST_TRUE: BindParameter[bool] = const(True) 

35"""A constant SQL expression `TRUE`. This will render as a literal `TRUE` in SQL queries.""" 

36 

37CONST_FALSE: BindParameter[bool] = const(False) 

38"""A constant SQL expression `FALSE`. This will render as a literal `FALSE` in SQL queries.""" 

39 

40CONST_NONE: BindParameter[None] = const(None) 

41"""A constant SQL expression `NULL`. This will render as a literal `NULL` in SQL queries.""" 

42 

43 

44def until_filter( 

45 until_time: BindParameter[datetime] | None = None, 

46 until_transaction_id: BindParameter[int] | None = None, 

47 until_inclusive: bool = True, 

48 transaction_time: QueryableAttribute = Transaction.time, 

49) -> ColumnExpressionArgument[bool]: 

50 """ 

51 Create a filter condition for transactions up to a given time or transaction. 

52 

53 Only one of `until_time` or `until_transaction_id` may be specified. 

54 """ 

55 

56 assert not (until_time is not None and until_transaction_id is not None), ( 

57 "Cannot filter by both until_time and until_transaction_id." 

58 ) 

59 

60 match (until_time, until_transaction_id, until_inclusive): 

61 case (BindParameter(), None, True): 

62 return transaction_time <= until_time 

63 case (BindParameter(), None, False): 63 ↛ 64line 63 didn't jump to line 64 because the pattern on line 63 never matched

64 return transaction_time < until_time 

65 case (None, BindParameter(), True): 

66 return ( 

67 transaction_time 

68 <= select(Transaction.time) 

69 .where(Transaction.id == until_transaction_id) 

70 .scalar_subquery() 

71 ) 

72 case (None, BindParameter(), False): 72 ↛ 73line 72 didn't jump to line 73 because the pattern on line 72 never matched

73 return ( 

74 transaction_time 

75 < select(Transaction.time) 

76 .where(Transaction.id == until_transaction_id) 

77 .scalar_subquery() 

78 ) 

79 

80 return CONST_TRUE 

81 

82 

83def after_filter( 

84 after_time: BindParameter[datetime] | None = None, 

85 after_transaction_id: BindParameter[int] | None = None, 

86 after_inclusive: bool = True, 

87 transaction_time: QueryableAttribute = Transaction.time, 

88) -> ColumnExpressionArgument[bool]: 

89 """ 

90 Create a filter condition for transactions after a given time or transaction. 

91 

92 Only one of `after_time` or `after_transaction_id` may be specified. 

93 """ 

94 

95 assert not (after_time is not None and after_transaction_id is not None), ( 

96 "Cannot filter by both after_time and after_transaction_id." 

97 ) 

98 

99 match (after_time, after_transaction_id, after_inclusive): 

100 case (BindParameter(), None, True): 100 ↛ 101line 100 didn't jump to line 101 because the pattern on line 100 never matched

101 return transaction_time >= after_time 

102 case (BindParameter(), None, False): 102 ↛ 103line 102 didn't jump to line 103 because the pattern on line 102 never matched

103 return transaction_time > after_time 

104 case (None, BindParameter(), True): 104 ↛ 105line 104 didn't jump to line 105 because the pattern on line 104 never matched

105 return ( 

106 transaction_time 

107 >= select(Transaction.time) 

108 .where(Transaction.id == after_transaction_id) 

109 .scalar_subquery() 

110 ) 

111 case (None, BindParameter(), False): 

112 return ( 

113 transaction_time 

114 > select(Transaction.time) 

115 .where(Transaction.id == after_transaction_id) 

116 .scalar_subquery() 

117 ) 

118 

119 return CONST_TRUE