Coverage for polar/customer_meter/endpoints.py: 72%
27 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from fastapi import Depends, Query 1a
3from polar.customer.schemas.customer import CustomerID 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.organization.schemas import OrganizationID 1a
11from polar.postgres import AsyncSession, get_db_session 1a
12from polar.routing import APIRouter 1a
14from . import auth, sorting 1a
15from .schemas import CustomerMeter as CustomerMeterSchema 1a
16from .schemas import CustomerMeterID 1a
17from .service import customer_meter as customer_meter_service 1a
19router = APIRouter( 1a
20 prefix="/customer-meters",
21 tags=["customer_meters", APITag.public, APITag.mcp],
22)
25CustomerMeterNotFound = { 1a
26 "description": "Customer meter not found.",
27 "model": ResourceNotFound.schema(),
28}
31@router.get( 1a
32 "/",
33 summary="List Customer Meters",
34 response_model=ListResource[CustomerMeterSchema],
35)
36async def list( 1a
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 )
68 return ListResource.from_paginated_results(
69 [CustomerMeterSchema.model_validate(result) for result in results],
70 count,
71 pagination,
72 )
75@router.get( 1a
76 "/{id}",
77 summary="Get Customer Meter",
78 response_model=CustomerMeterSchema,
79 responses={404: CustomerMeterNotFound},
80)
81async def get( 1a
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)
89 if customer_meter is None:
90 raise ResourceNotFound()
92 return customer_meter