Coverage for polar/transaction/endpoints.py: 94%

1from typing import Annotated 

2 

3from fastapi import Depends, Query 

4from pydantic import UUID4 

5 

6from polar.account.service import account as account_service 

7from polar.exceptions import ResourceNotFound 

8from polar.kit.pagination import ListResource, PaginationParamsQuery 

9from polar.kit.sorting import Sorting, SortingGetter 

10from polar.models.transaction import TransactionType 

11from polar.openapi import APITag 

12from polar.postgres import AsyncReadSession, get_db_read_session 

13from polar.routing import APIRouter 

14from polar.transaction import ( 

15 auth as transactions_auth, 

16) 

17 

18from .schemas import Transaction, TransactionDetails, TransactionsSummary 

19from .service.transaction import TransactionSortProperty 

20from .service.transaction import transaction as transaction_service 

21 

22router = APIRouter(prefix="/transactions", tags=["transactions", APITag.private]) 

23 

24 

25SearchSorting = Annotated[ 

26 list[Sorting[TransactionSortProperty]], 

27 Depends(SortingGetter(TransactionSortProperty, ["-created_at"])), 

28] 

29 

30 

31@router.get("/search", response_model=ListResource[Transaction]) 

32async def search_transactions( 

33 pagination: PaginationParamsQuery, 

34 sorting: SearchSorting, 

35 auth_subject: transactions_auth.TransactionsRead, 

36 type: TransactionType | None = Query(None), 

37 account_id: UUID4 | None = Query(None), 

38 payment_customer_id: UUID4 | None = Query(None), 

39 payment_organization_id: UUID4 | None = Query(None), 

40 payment_user_id: UUID4 | None = Query(None), 

41 exclude_platform_fees: bool = Query(False), 

42 session: AsyncReadSession = Depends(get_db_read_session), 

43) -> ListResource[Transaction]: 

44 results, count = await transaction_service.search( 

45 session, 

46 auth_subject.subject, 

47 type=type, 

48 account_id=account_id, 

49 payment_customer_id=payment_customer_id, 

50 payment_organization_id=payment_organization_id, 

51 payment_user_id=payment_user_id, 

52 exclude_platform_fees=exclude_platform_fees, 

53 pagination=pagination, 

54 sorting=sorting, 

55 ) 

56 

57 return ListResource.from_paginated_results( 

58 [Transaction.model_validate(result) for result in results], 

59 count, 

60 pagination, 

61 ) 

62 

63 

64@router.get("/lookup", response_model=TransactionDetails) 

65async def lookup_transaction( 

66 transaction_id: UUID4, 

67 auth_subject: transactions_auth.TransactionsRead, 

68 session: AsyncReadSession = Depends(get_db_read_session), 

69) -> TransactionDetails: 

70 return TransactionDetails.model_validate( 

71 await transaction_service.lookup(session, transaction_id, auth_subject.subject) 

72 ) 

73 

74 

75@router.get("/summary", response_model=TransactionsSummary) 

76async def get_summary( 

77 auth_subject: transactions_auth.TransactionsRead, 

78 account_id: UUID4, 

79 session: AsyncReadSession = Depends(get_db_read_session), 

80) -> TransactionsSummary: 

81 account = await account_service.get(session, auth_subject, account_id) 

82 if account is None: 82 ↛ 85line 82 didn't jump to line 85 because the condition on line 82 was always true

83 raise ResourceNotFound() 

84 

85 return await transaction_service.get_summary(session, account)