Coverage for polar/transaction/service/base.py: 69%

1import itertools 

2 

3from sqlalchemy import select 

4 

5from polar.exceptions import PolarError 

6from polar.kit.services import ResourceServiceReader 

7from polar.models import Transaction 

8from polar.models.transaction import TransactionType 

9from polar.postgres import AsyncSession 

10 

11 

12class BaseTransactionServiceError(PolarError): ... 12 ↛ 15line 12 didn't jump to line 15 because

13 

14 

15class BaseTransactionService(ResourceServiceReader[Transaction]): 

16 async def _get_balance_transactions_for_payment( 

17 self, session: AsyncSession, *, payment_transaction: Transaction 

18 ) -> list[tuple[Transaction, Transaction]]: 

19 statement = ( 

20 select(Transaction) 

21 .where( 

22 Transaction.type == TransactionType.balance, 

23 Transaction.payment_transaction_id == payment_transaction.id, 

24 ) 

25 .order_by( 

26 Transaction.balance_correlation_key, 

27 Transaction.account_id.nulls_first(), 

28 ) 

29 ) 

30 

31 result = await session.execute(statement) 

32 transactions = list(result.scalars().all()) 

33 return [ 

34 (t1, t2) 

35 for _, (t1, t2) in itertools.groupby( 

36 transactions, key=lambda t: t.balance_correlation_key 

37 ) 

38 ]