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

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