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

14 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1import itertools 1a

2 

3from sqlalchemy import select 1a

4 

5from polar.exceptions import PolarError 1a

6from polar.kit.services import ResourceServiceReader 1a

7from polar.models import Transaction 1a

8from polar.models.transaction import TransactionType 1a

9from polar.postgres import AsyncSession 1a

10 

11 

12class BaseTransactionServiceError(PolarError): ... 1a

13 

14 

15class BaseTransactionService(ResourceServiceReader[Transaction]): 1a

16 async def _get_balance_transactions_for_payment( 1a

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 ]