Coverage for dibbler / queries / affected_users.py: 66%

1from datetime import datetime 

2 

3from sqlalchemy import BindParameter, select, or_ 

4from sqlalchemy.orm import Session 

5 

6from dibbler.models import Transaction, TransactionType, User 

7from dibbler.queries.query_helpers import until_filter, after_filter 

8 

9def affected_users( 

10 sql_session: Session, 

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

12 until_transaction: BindParameter[Transaction] | Transaction | None = None, 

13 until_inclusive: bool = True, 

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

15 after_transaction: Transaction | None = None, 

16 after_inclusive: bool = True, 

17) -> set[User]: 

18 """ 

19 Get all users where attributes were affected over a given interval. 

20 """ 

21 

22 if isinstance(until_time, datetime): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

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

24 

25 if isinstance(until_transaction, Transaction): 

26 if until_transaction.id is None: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

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

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

29 else: 

30 until_transaction_id = None 

31 

32 if not (after_time is None or after_transaction is None): 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

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

34 

35 if isinstance(after_time, datetime): 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

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

37 

38 if isinstance(after_transaction, Transaction): 

39 if after_transaction.id is None: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

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

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

42 else: 

43 after_transaction_id = None 

44 

45 if after_time is not None and until_time is not None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 assert isinstance(after_time.value, datetime) 

47 assert isinstance(until_time.value, datetime) 

48 

49 if after_time.value > until_time.value: 

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

51 

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

53 assert after_transaction.time is not None 

54 assert until_transaction.time is not None 

55 

56 if after_transaction.time > until_transaction.time: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

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

58 

59 result = sql_session.scalars( 

60 select(User) 

61 .distinct() 

62 .join(Transaction, or_(User.id == Transaction.user_id, User.id == Transaction.transfer_user_id)) 

63 .where( 

64 Transaction.type_.in_( 

65 [ 

66 TransactionType.ADD_PRODUCT.as_literal_column(), 

67 TransactionType.ADJUST_BALANCE.as_literal_column(), 

68 TransactionType.BUY_PRODUCT.as_literal_column(), 

69 TransactionType.TRANSFER.as_literal_column(), 

70 ] 

71 ), 

72 until_filter( 

73 until_time=until_time, 

74 until_transaction_id=until_transaction_id, 

75 until_inclusive=until_inclusive, 

76 ), 

77 after_filter( 

78 after_time=after_time, 

79 after_transaction_id=after_transaction_id, 

80 after_inclusive=after_inclusive, 

81 ), 

82 ) 

83 .order_by(Transaction.time.desc()) 

84 ).all() 

85 

86 return set(result)