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

1from datetime import datetime 

2 

3from sqlalchemy.orm import Session 

4 

5from dibbler.models import ( 

6 Product, 

7 Transaction, 

8 User, 

9) 

10 

11 

12def joint_buy_product( 

13 sql_session: Session, 

14 product: Product, 

15 product_count: int, 

16 instigator: User, 

17 users: list[User], 

18 time: datetime | None = None, 

19 message: str | None = None, 

20) -> list[Transaction]: 

21 """ 

22 Create buy product transactions for multiple users at once. 

23 """ 

24 

25 if product.id is None: 

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

27 

28 if instigator.id is None: 

29 raise ValueError("Instigator must be persisted in the database.") 

30 

31 if len(users) == 0: 

32 raise ValueError("At least bying one user must be specified.") 

33 

34 if any(user.id is None for user in users): 

35 raise ValueError("All users must be persisted in the database.") 

36 

37 if instigator not in users: 

38 raise ValueError("Instigator must be in the list of users buying the product.") 

39 

40 if product_count <= 0: 

41 raise ValueError("Product count must be positive.") 

42 

43 # TODO: verify time is not behind last transaction's time 

44 

45 joint_transaction = Transaction.joint( 

46 user_id=instigator.id, 

47 product_id=product.id, 

48 product_count=product_count, 

49 time=time, 

50 message=message, 

51 ) 

52 sql_session.add(joint_transaction) 

53 sql_session.flush() # Ensure joint_transaction gets an ID 

54 

55 transactions = [joint_transaction] 

56 

57 for user in users: 

58 buy_transaction = Transaction.joint_buy_product( 

59 user_id=user.id, 

60 joint_transaction_id=joint_transaction.id, 

61 time=time, 

62 message=message, 

63 ) 

64 sql_session.add(buy_transaction) 

65 transactions.append(buy_transaction) 

66 

67 sql_session.commit() 

68 return transactions