Coverage for polar/processor_transaction/repository.py: 47%

1from collections.abc import Sequence 1 ctx1a

2 

3from polar.kit.repository.base import RepositoryBase 1 ctx1a

4from polar.models import ProcessorTransaction 1 ctx1a

5from polar.models.processor_transaction import Processor 1 ctx1a

6 

7 

8class ProcessorTransactionRepository(RepositoryBase[ProcessorTransaction]): 1 ctx1a

9 model = ProcessorTransaction 1 ctx1a

10 

11 async def get_latest_by_processor( 1 ctx1a

12 self, processor: Processor 

13 ) -> Sequence[ProcessorTransaction]: 

14 latest_timestamp_statement = ( 

15 self.get_base_statement() 

16 .where(ProcessorTransaction.processor == processor) 

17 .order_by(ProcessorTransaction.timestamp.desc()) 

18 .limit(1) 

19 ) 

20 latest_transaction = await self.get_one_or_none(latest_timestamp_statement) 

21 

22 if latest_transaction is None: 

23 return [] 

24 

25 statement = self.get_base_statement().where( 

26 ProcessorTransaction.processor == processor, 

27 ProcessorTransaction.timestamp == latest_transaction.timestamp, 

28 ) 

29 return await self.get_all(statement)