Coverage for polar/transaction/repository.py: 63%
35 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from collections.abc import Sequence 1a
2from uuid import UUID 1a
4from sqlalchemy import Select 1a
5from sqlalchemy.orm import selectinload 1a
7from polar.kit.repository import ( 1a
8 RepositoryBase,
9 RepositorySoftDeletionIDMixin,
10 RepositorySoftDeletionMixin,
11)
12from polar.models import Order, Transaction 1a
13from polar.models.transaction import TransactionType 1a
16class TransactionRepository( 1a
17 RepositorySoftDeletionIDMixin[Transaction, UUID],
18 RepositorySoftDeletionMixin[Transaction],
19 RepositoryBase[Transaction],
20):
21 model = Transaction 1a
23 async def get_all_paid_transactions_by_payout( 1a
24 self, payout_transaction_id: UUID
25 ) -> Sequence[Transaction]:
26 statement = self.get_paid_transactions_statement(payout_transaction_id)
27 return await self.get_all(statement)
29 def get_paid_transactions_statement( 1a
30 self, payout_transaction_id: UUID
31 ) -> Select[tuple[Transaction]]:
32 return (
33 self.get_base_statement()
34 .where(
35 Transaction.payout_transaction_id == payout_transaction_id,
36 )
37 .order_by(Transaction.created_at)
38 .options(
39 # Order
40 selectinload(Transaction.order).joinedload(Order.product),
41 # Pledge
42 selectinload(Transaction.pledge),
43 )
44 )
47class PaymentTransactionRepository(TransactionRepository): 1a
48 def get_base_statement( 1a
49 self, *, include_deleted: bool = False
50 ) -> Select[tuple[Transaction]]:
51 return (
52 super()
53 .get_base_statement(include_deleted=include_deleted)
54 .where(Transaction.type == TransactionType.payment)
55 )
58class BalanceTransactionRepository(TransactionRepository): 1a
59 async def get_all_unpaid_by_account(self, account: UUID) -> Sequence[Transaction]: 1a
60 statement = (
61 self.get_base_statement()
62 .where(
63 Transaction.type == TransactionType.balance,
64 Transaction.account_id == account,
65 Transaction.payout_transaction_id.is_(None),
66 )
67 .options(
68 selectinload(Transaction.balance_reversal_transaction),
69 selectinload(Transaction.balance_reversal_transactions),
70 selectinload(Transaction.payment_transaction),
71 )
72 )
73 return await self.get_all(statement)
75 def get_base_statement( 1a
76 self, *, include_deleted: bool = False
77 ) -> Select[tuple[Transaction]]:
78 return (
79 super()
80 .get_base_statement(include_deleted=include_deleted)
81 .where(Transaction.type == TransactionType.balance)
82 )
85class RefundTransactionRepository(TransactionRepository): 1a
86 async def get_by_refund_id(self, refund_id: str) -> Transaction | None: 1a
87 statement = self.get_base_statement().where(Transaction.refund_id == refund_id)
88 return await self.get_one_or_none(statement)
90 def get_base_statement( 1a
91 self, *, include_deleted: bool = False
92 ) -> Select[tuple[Transaction]]:
93 return (
94 super()
95 .get_base_statement(include_deleted=include_deleted)
96 .where(Transaction.type == TransactionType.refund)
97 )
100class PayoutTransactionRepository(TransactionRepository): 1a
101 async def get_by_payout_id(self, payout_id: UUID) -> Transaction | None: 1a
102 statement = self.get_base_statement().where(Transaction.payout_id == payout_id)
103 return await self.get_one_or_none(statement)
105 def get_base_statement( 1a
106 self, *, include_deleted: bool = False
107 ) -> Select[tuple[Transaction]]:
108 return (
109 super()
110 .get_base_statement(include_deleted=include_deleted)
111 .where(Transaction.type == TransactionType.payout)
112 )