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

1from fastapi import Depends, Query 

2 

3from polar.customer_meter.schemas import CustomerMeterID 

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.postgres import AsyncSession, get_db_session 

11from polar.routing import APIRouter 

12 

13from .. import auth 

14from ..schemas.customer_meter import CustomerCustomerMeter as CustomerMeterSchema 

15from ..service.customer_meter import customer_meter as customer_meter_service 

16from ..sorting import customer_meter as sorting 

17 

18router = APIRouter( 

19 prefix="/meters", 

20 tags=["customer_meters", APITag.public], 

21) 

22 

23 

24CustomerMeterNotFound = { 

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

26 "model": ResourceNotFound.schema(), 

27} 

28 

29 

30@router.get( 

31 "/", 

32 summary="List Meters", 

33 response_model=ListResource[CustomerMeterSchema], 

34) 

35async def list( 

36 auth_subject: auth.CustomerPortalRead, 

37 pagination: PaginationParamsQuery, 

38 sorting: sorting.ListSorting, 

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

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

41 ), 

42 query: str | None = Query(None, description="Filter by meter name."), 

43 session: AsyncSession = Depends(get_db_session), 

44) -> ListResource[CustomerMeterSchema]: 

45 """List meters of the authenticated customer.""" 

46 results, count = await customer_meter_service.list( 

47 session, 

48 auth_subject, 

49 meter_id=meter_id, 

50 query=query, 

51 pagination=pagination, 

52 sorting=sorting, 

53 ) 

54 

55 return ListResource.from_paginated_results( 

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

57 count, 

58 pagination, 

59 ) 

60 

61 

62@router.get( 

63 "/{id}", 

64 summary="Get Customer Meter", 

65 response_model=CustomerMeterSchema, 

66 responses={404: CustomerMeterNotFound}, 

67) 

68async def get( 

69 id: CustomerMeterID, 

70 auth_subject: auth.CustomerPortalRead, 

71 session: AsyncSession = Depends(get_db_session), 

72) -> CustomerMeter: 

73 """Get a meter by ID for the authenticated customer.""" 

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

75 

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

77 raise ResourceNotFound() 

78 

79 return customer_meter