Coverage for polar/customer_portal/endpoints/customer_meter.py: 71%
26 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from fastapi import Depends, Query 1a
3from polar.customer_meter.schemas import CustomerMeterID 1a
4from polar.exceptions import ResourceNotFound 1a
5from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
6from polar.kit.schemas import MultipleQueryFilter 1a
7from polar.meter.schemas import MeterID 1a
8from polar.models import CustomerMeter 1a
9from polar.openapi import APITag 1a
10from polar.postgres import AsyncSession, get_db_session 1a
11from polar.routing import APIRouter 1a
13from .. import auth 1a
14from ..schemas.customer_meter import CustomerCustomerMeter as CustomerMeterSchema 1a
15from ..service.customer_meter import customer_meter as customer_meter_service 1a
16from ..sorting import customer_meter as sorting 1a
18router = APIRouter( 1a
19 prefix="/meters",
20 tags=["customer_meters", APITag.public],
21)
24CustomerMeterNotFound = { 1a
25 "description": "Customer meter not found.",
26 "model": ResourceNotFound.schema(),
27}
30@router.get( 1a
31 "/",
32 summary="List Meters",
33 response_model=ListResource[CustomerMeterSchema],
34)
35async def list( 1a
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 )
55 return ListResource.from_paginated_results(
56 [CustomerMeterSchema.model_validate(result) for result in results],
57 count,
58 pagination,
59 )
62@router.get( 1a
63 "/{id}",
64 summary="Get Customer Meter",
65 response_model=CustomerMeterSchema,
66 responses={404: CustomerMeterNotFound},
67)
68async def get( 1a
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)
76 if customer_meter is None:
77 raise ResourceNotFound()
79 return customer_meter