Coverage for dibbler / queries / update_cache.py: 93%

1from sqlalchemy import insert, select 

2from sqlalchemy.orm import Session 

3 

4from dibbler.models import LastCacheTransaction, ProductCache, Transaction, UserCache 

5from dibbler.queries.affected_products import affected_products 

6from dibbler.queries.affected_users import affected_users 

7from dibbler.queries.product_price import product_price 

8from dibbler.queries.product_stock import product_stock 

9from dibbler.queries.user_balance import user_balance 

10 

11 

12def update_cache( 

13 sql_session: Session, 

14 use_cache: bool = True, 

15) -> None: 

16 """ 

17 Update the cache used for searching products. 

18 

19 If `use_cache` is False, the cache will be rebuilt from scratch. 

20 """ 

21 

22 last_transaction = sql_session.scalars( 

23 select(Transaction).order_by(Transaction.time.desc()).limit(1) 

24 ).one_or_none() 

25 

26 print(last_transaction) 

27 

28 if last_transaction is None: 

29 # No transactions exist, nothing to update 

30 return 

31 

32 if use_cache: 32 ↛ 42line 32 didn't jump to line 42 because the condition on line 32 was always true

33 last_cache_transaction = sql_session.scalars( 

34 select(LastCacheTransaction) 

35 .join(Transaction, LastCacheTransaction.transaction_id == Transaction.id) 

36 .order_by(Transaction.time.desc()) 

37 .limit(1) 

38 ).one_or_none() 

39 if last_cache_transaction is not None: 

40 last_cache_transaction = last_cache_transaction.transaction 

41 else: 

42 last_cache_transaction = None 

43 

44 if last_cache_transaction is not None and last_cache_transaction.id == last_transaction.id: 

45 # Cache is already up to date 

46 return 

47 

48 users = affected_users( 

49 sql_session, 

50 after_transaction=last_cache_transaction, 

51 after_inclusive=False, 

52 until_transaction=last_transaction, 

53 ) 

54 products = affected_products( 

55 sql_session, 

56 after_transaction=last_cache_transaction, 

57 after_inclusive=False, 

58 until_transaction=last_transaction, 

59 ) 

60 

61 user_balances = {} 

62 for user in users: 

63 x = user_balance( 

64 sql_session, 

65 user, 

66 use_cache=use_cache, 

67 until_transaction=last_transaction, 

68 ) 

69 user_balances[user.id] = x 

70 

71 product_stocks = {} 

72 product_prices = {} 

73 for product in products: 

74 product_stocks[product.id] = product_stock( 

75 sql_session, 

76 product, 

77 use_cache=use_cache, 

78 until_transaction=last_transaction, 

79 ) 

80 product_prices[product.id] = product_price( 

81 sql_session, 

82 product, 

83 use_cache=use_cache, 

84 until_transaction=last_transaction, 

85 ) 

86 

87 next_cache_transaction = LastCacheTransaction(transaction_id=last_transaction.id) 

88 sql_session.add(next_cache_transaction) 

89 sql_session.flush() 

90 

91 if not len(users) == 0: 91 ↛ 104line 91 didn't jump to line 104 because the condition on line 91 was always true

92 sql_session.execute( 

93 insert(UserCache), 

94 [ 

95 { 

96 "user_id": user.id, 

97 "balance": user_balances[user.id], 

98 "last_cache_transaction_id": next_cache_transaction.id, 

99 } 

100 for user in users 

101 ], 

102 ) 

103 

104 if not len(products) == 0: 104 ↛ 118line 104 didn't jump to line 118 because the condition on line 104 was always true

105 sql_session.execute( 

106 insert(ProductCache), 

107 [ 

108 { 

109 "product_id": product.id, 

110 "stock": product_stocks[product.id], 

111 "price": product_prices[product.id], 

112 "last_cache_transaction_id": next_cache_transaction.id, 

113 } 

114 for product in products 

115 ], 

116 ) 

117 

118 sql_session.commit()