Coverage for dibbler / queries / transaction_log.py: 100%

1from datetime import datetime 

2 

3from sqlalchemy import BindParameter, select 

4from sqlalchemy.orm import Session 

5 

6from dibbler.models import ( 

7 Product, 

8 Transaction, 

9 TransactionType, 

10 User, 

11) 

12 

13 

14# TODO: should this include full joint transactions that involve a user? 

15# TODO: should this involve throw-away transactions that affects a user? 

16def transaction_log( 

17 sql_session: Session, 

18 user: User | None = None, 

19 product: Product | None = None, 

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

21 until_transaction: Transaction | None = None, 

22 until_inclusive: bool = True, 

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

24 after_transaction: Transaction | None = None, 

25 after_inlcusive: bool = True, 

26 transaction_type: list[TransactionType] | None = None, 

27 negate_transaction_type_filter: bool = False, 

28 limit: int | None = None, 

29) -> list[Transaction]: 

30 """ 

31 Retrieve the transaction log, optionally filtered. 

32 

33 Only one of `user` or `product` may be specified. 

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

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

36 

37 The after and after filters are inclusive by default. 

38 """ 

39 

40 if not (user is None or product is None): 

41 raise ValueError("Cannot filter by both user and product.") 

42 

43 if isinstance(user, User): 

44 if user.id is None: 

45 raise ValueError("User must be persisted in the database.") 

46 user_id = BindParameter("user_id", value=user.id) 

47 else: 

48 user_id = None 

49 

50 if isinstance(product, Product): 

51 if product.id is None: 

52 raise ValueError("Product must be persisted in the database.") 

53 product_id = BindParameter("product_id", value=product.id) 

54 else: 

55 product_id = None 

56 

57 if not (until_time is None or until_transaction is None): 

58 raise ValueError("Cannot filter by both after_time and after_transaction_id.") 

59 

60 if isinstance(until_time, datetime): 

61 until_time = BindParameter("until_time", value=until_time) 

62 

63 if isinstance(until_transaction, Transaction): 

64 if until_transaction.id is None: 

65 raise ValueError("until_transaction must be persisted in the database.") 

66 until_transaction_id = BindParameter("until_transaction_id", value=until_transaction.id) 

67 else: 

68 until_transaction_id = None 

69 

70 if not (after_time is None or after_transaction is None): 

71 raise ValueError("Cannot filter by both after_time and after_transaction_id.") 

72 

73 if isinstance(after_time, datetime): 

74 after_time = BindParameter("after_time", value=after_time) 

75 

76 if isinstance(after_transaction, Transaction): 

77 if after_transaction.id is None: 

78 raise ValueError("after_transaction must be persisted in the database.") 

79 after_transaction_id = BindParameter("after_transaction_id", value=after_transaction.id) 

80 else: 

81 after_transaction_id = None 

82 

83 if after_time is not None and until_time is not None: 

84 assert isinstance(after_time.value, datetime) 

85 assert isinstance(until_time.value, datetime) 

86 

87 if after_time.value > until_time.value: 

88 raise ValueError("after_time cannot be after until_time.") 

89 

90 if after_transaction is not None and until_transaction is not None: 

91 assert after_transaction.time is not None 

92 assert until_transaction.time is not None 

93 

94 if after_transaction.time > until_transaction.time: 

95 raise ValueError("after_transaction cannot be after until_transaction.") 

96 

97 if limit is not None and limit <= 0: 

98 raise ValueError("Limit must be positive.") 

99 

100 query = select(Transaction) 

101 if user is not None: 

102 query = query.where(Transaction.user_id == user_id) 

103 if product is not None: 

104 query = query.where(Transaction.product_id == product_id) 

105 

106 match (until_time, until_transaction_id, until_inclusive): 

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

108 query = query.where(Transaction.time <= until_time) 

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

110 query = query.where(Transaction.time < until_time) 

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

112 query = query.where(Transaction.id <= until_transaction_id) 

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

114 query = query.where(Transaction.id < until_transaction_id) 

115 case _: 

116 pass 

117 

118 match (after_time, after_transaction_id, after_inlcusive): 

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

120 query = query.where(Transaction.time >= after_time) 

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

122 query = query.where(Transaction.time > after_time) 

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

124 query = query.where(Transaction.id >= after_transaction_id) 

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

126 query = query.where(Transaction.id > after_transaction_id) 

127 case _: 

128 pass 

129 

130 if transaction_type is not None: 

131 if negate_transaction_type_filter: 

132 query = query.where(~Transaction.type_.in_(transaction_type)) 

133 else: 

134 query = query.where(Transaction.type_.in_(transaction_type)) 

135 

136 if limit is not None: 

137 query = query.limit(limit) 

138 

139 query = query.order_by(Transaction.time.asc(), Transaction.id.asc()) 

140 result = sql_session.scalars(query).all() 

141 

142 return list(result)