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

1from fastapi import Depends, Query 

2 

3from polar.customer.schemas.customer import CustomerID 

4from polar.exceptions import ResourceNotFound 

5from polar.kit.pagination import ListResource, PaginationParamsQuery 

6from polar.kit.schemas import MultipleQueryFilter 

7from polar.meter.schemas import MeterID 

8from polar.models import CustomerMeter 

9from polar.openapi import APITag 

10from polar.organization.schemas import OrganizationID 

11from polar.postgres import AsyncSession, get_db_session 

12from polar.routing import APIRouter 

13 

14from . import auth, sorting 

15from .schemas import CustomerMeter as CustomerMeterSchema 

16from .schemas import CustomerMeterID 

17from .service import customer_meter as customer_meter_service 

18 

19router = APIRouter( 

20 prefix="/customer-meters", 

21 tags=["customer_meters", APITag.public, APITag.mcp], 

22) 

23 

24 

25CustomerMeterNotFound = { 

26 "description": "Customer meter not found.", 

27 "model": ResourceNotFound.schema(), 

28} 

29 

30 

31@router.get( 

32 "/", 

33 summary="List Customer Meters", 

34 response_model=ListResource[CustomerMeterSchema], 

35) 

36async def list( 

37 auth_subject: auth.CustomerMeterRead, 

38 pagination: PaginationParamsQuery, 

39 sorting: sorting.ListSorting, 

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

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

42 ), 

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

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

45 ), 

46 external_customer_id: MultipleQueryFilter[str] | None = Query( 

47 None, 

48 title="ExternalCustomerID Filter", 

49 description="Filter by external customer ID.", 

50 ), 

51 meter_id: MultipleQueryFilter[MeterID] | None = Query( 

52 None, title="MeterID Filter", description="Filter by meter ID." 

53 ), 

54 session: AsyncSession = Depends(get_db_session), 

55) -> ListResource[CustomerMeterSchema]: 

56 """List customer meters.""" 

57 results, count = await customer_meter_service.list( 

58 session, 

59 auth_subject, 

60 organization_id=organization_id, 

61 customer_id=customer_id, 

62 external_customer_id=external_customer_id, 

63 meter_id=meter_id, 

64 pagination=pagination, 

65 sorting=sorting, 

66 ) 

67 

68 return ListResource.from_paginated_results( 

69 [CustomerMeterSchema.model_validate(result) for result in results], 

70 count, 

71 pagination, 

72 ) 

73 

74 

75@router.get( 

76 "/{id}", 

77 summary="Get Customer Meter", 

78 response_model=CustomerMeterSchema, 

79 responses={404: CustomerMeterNotFound}, 

80) 

81async def get( 

82 id: CustomerMeterID, 

83 auth_subject: auth.CustomerMeterRead, 

84 session: AsyncSession = Depends(get_db_session), 

85) -> CustomerMeter: 

86 """Get a customer meter by ID.""" 

87 customer_meter = await customer_meter_service.get(session, auth_subject, id) 

88 

89 if customer_meter is None: 89 ↛ 92line 89 didn't jump to line 92 because the condition on line 89 was always true

90 raise ResourceNotFound() 

91 

92 return customer_meter