Coverage for polar/member/endpoints.py: 58%
22 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from uuid import UUID 1a
3from fastapi import Depends, Query 1a
5from polar.exceptions import ResourceNotFound 1a
6from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
7from polar.openapi import APITag 1a
8from polar.postgres import AsyncReadSession, get_db_read_session 1a
9from polar.routing import APIRouter 1a
11from . import auth, sorting 1a
12from .schemas import Member 1a
13from .service import member_service 1a
15router = APIRouter( 1a
16 prefix="/members",
17 tags=["members", APITag.public, APITag.mcp],
18)
20MemberNotFound = { 1a
21 "description": "Member not found.",
22 "model": ResourceNotFound.schema(),
23}
26@router.get( 1a
27 "/",
28 summary="List Members",
29 response_model=ListResource[Member],
30)
31async def list_members( 1a
32 auth_subject: auth.MemberRead,
33 pagination: PaginationParamsQuery,
34 sorting: sorting.ListSorting,
35 customer_id: str | None = Query(None, description="Filter by customer ID."),
36 session: AsyncReadSession = Depends(get_db_read_session),
37) -> ListResource[Member]:
38 """List members with optional customer ID filter."""
39 parsed_customer_id = None
40 if customer_id is not None:
41 try:
42 parsed_customer_id = UUID(customer_id)
43 except ValueError:
44 raise ResourceNotFound("Invalid customer ID format")
46 results, count = await member_service.list(
47 session,
48 auth_subject,
49 customer_id=parsed_customer_id,
50 pagination=pagination,
51 sorting=sorting,
52 )
54 return ListResource.from_paginated_results(
55 [Member.model_validate(member) for member in results],
56 count,
57 pagination,
58 )