Coverage for polar/payment/endpoints.py: 93%

1from fastapi import Depends, Query 

2from pydantic.types import UUID4 

3 

4from polar.exceptions import ResourceNotFound 

5from polar.kit.pagination import ListResource, PaginationParamsQuery 

6from polar.kit.schemas import MultipleQueryFilter 

7from polar.models import Payment 

8from polar.models.payment import PaymentStatus 

9from polar.openapi import APITag 

10from polar.organization.schemas import OrganizationID 

11from polar.postgres import AsyncReadSession, get_db_read_session 

12from polar.routing import APIRouter 

13 

14from . import auth, sorting 

15from .schemas import Payment as PaymentSchema 

16from .schemas import PaymentAdapter, PaymentID 

17from .service import payment as payment_service 

18 

19router = APIRouter(prefix="/payments", tags=["payments", APITag.public, APITag.mcp]) 

20 

21 

22PaymentNotFound = { 

23 "description": "Payment not found.", 

24 "model": ResourceNotFound.schema(), 

25} 

26 

27 

28@router.get("/", summary="List Payments", response_model=ListResource[PaymentSchema]) 

29async def list( 

30 auth_subject: auth.PaymentRead, 

31 pagination: PaginationParamsQuery, 

32 sorting: sorting.ListSorting, 

33 organization_id: MultipleQueryFilter[OrganizationID] | None = Query( 

34 None, title="OrganizationID Filter", description="Filter by organization ID." 

35 ), 

36 checkout_id: MultipleQueryFilter[UUID4] | None = Query( 

37 None, title="CheckoutID Filter", description="Filter by checkout ID." 

38 ), 

39 order_id: MultipleQueryFilter[UUID4] | None = Query( 

40 None, title="OrderID Filter", description="Filter by order ID." 

41 ), 

42 status: MultipleQueryFilter[PaymentStatus] | None = Query( 

43 None, title="Status Filter", description="Filter by payment status." 

44 ), 

45 method: MultipleQueryFilter[str] | None = Query( 

46 None, title="Method Filter", description="Filter by payment method." 

47 ), 

48 customer_email: MultipleQueryFilter[str] | None = Query( 

49 None, title="CustomerEmail Filter", description="Filter by customer email." 

50 ), 

51 session: AsyncReadSession = Depends(get_db_read_session), 

52) -> ListResource[PaymentSchema]: 

53 """List payments.""" 

54 results, count = await payment_service.list( 

55 session, 

56 auth_subject, 

57 organization_id=organization_id, 

58 checkout_id=checkout_id, 

59 order_id=order_id, 

60 status=status, 

61 method=method, 

62 customer_email=customer_email, 

63 pagination=pagination, 

64 sorting=sorting, 

65 ) 

66 

67 return ListResource.from_paginated_results( 

68 [PaymentAdapter.validate_python(result) for result in results], 

69 count, 

70 pagination, 

71 ) 

72 

73 

74@router.get( 

75 "/{id}", 

76 summary="Get Payment", 

77 response_model=PaymentSchema, 

78 responses={404: PaymentNotFound}, 

79) 

80async def get( 

81 id: PaymentID, 

82 auth_subject: auth.PaymentRead, 

83 session: AsyncReadSession = Depends(get_db_read_session), 

84) -> Payment: 

85 """Get a payment by ID.""" 

86 payment = await payment_service.get(session, auth_subject, id) 

87 

88 if payment is None: 88 ↛ 91line 88 didn't jump to line 91 because the condition on line 88 was always true

89 raise ResourceNotFound() 

90 

91 return payment