Coverage for polar/refund/endpoints.py: 100%

1from fastapi import Depends, Query 

2 

3from polar.customer.schemas.customer import CustomerID 

4from polar.kit.pagination import ListResource, PaginationParamsQuery 

5from polar.kit.schemas import MultipleQueryFilter 

6from polar.models import Refund 

7from polar.openapi import APITag 

8from polar.order.schemas import OrderID 

9from polar.organization.schemas import OrganizationID 

10from polar.postgres import AsyncSession, get_db_session 

11from polar.routing import APIRouter 

12from polar.subscription.schemas import SubscriptionID 

13 

14from . import auth 

15from .schemas import Refund as RefundSchema 

16from .schemas import RefundCreate, RefundID 

17from .service import RefundAmountTooHigh, RefundedAlready 

18from .service import refund as refund_service 

19from .sorting import RefundListSorting 

20 

21router = APIRouter(prefix="/refunds", tags=["refunds", APITag.public]) 

22 

23 

24@router.get("/", summary="List Refunds", response_model=ListResource[RefundSchema]) 

25async def list( 

26 pagination: PaginationParamsQuery, 

27 sorting: RefundListSorting, 

28 auth_subject: auth.RefundsRead, 

29 id: MultipleQueryFilter[RefundID] | None = Query( 

30 None, title="RefundID Filter", description="Filter by refund ID." 

31 ), 

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

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

34 ), 

35 order_id: MultipleQueryFilter[OrderID] | None = Query( 

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

37 ), 

38 subscription_id: MultipleQueryFilter[SubscriptionID] | None = Query( 

39 None, title="SubscriptionID Filter", description="Filter by subscription ID." 

40 ), 

41 customer_id: MultipleQueryFilter[CustomerID] | None = Query( 

42 None, title="CustomerID Filter", description="Filter by customer ID." 

43 ), 

44 succeeded: bool | None = Query( 

45 None, 

46 title="RefundStatus Filter", 

47 description="Filter by `succeeded`.", 

48 ), 

49 session: AsyncSession = Depends(get_db_session), 

50) -> ListResource[RefundSchema]: 

51 """List products.""" 

52 results, count = await refund_service.get_list( 

53 session, 

54 auth_subject, 

55 id=id, 

56 organization_id=organization_id, 

57 order_id=order_id, 

58 subscription_id=subscription_id, 

59 customer_id=customer_id, 

60 succeeded=succeeded, 

61 pagination=pagination, 

62 sorting=sorting, 

63 ) 

64 

65 return ListResource.from_paginated_results( 

66 [RefundSchema.model_validate(result) for result in results], 

67 count, 

68 pagination, 

69 ) 

70 

71 

72@router.post( 

73 "/", 

74 summary="Create Refund", 

75 response_model=RefundSchema, 

76 responses={ 

77 201: {"description": "Refund created."}, 

78 400: { 

79 "description": "Refund amount exceeds remaining order balance.", 

80 "model": RefundAmountTooHigh.schema(), 

81 }, 

82 403: { 

83 "description": "Order is already fully refunded.", 

84 "model": RefundedAlready.schema(), 

85 }, 

86 }, 

87) 

88async def create( 

89 refund_create: RefundCreate, 

90 auth_subject: auth.RefundsWrite, 

91 session: AsyncSession = Depends(get_db_session), 

92) -> Refund: 

93 """Create a refund.""" 

94 return await refund_service.user_create( 

95 session, 

96 auth_subject, 

97 create_schema=refund_create, 

98 )