Coverage for polar/member/endpoints.py: 100%

1from uuid import UUID 

2 

3from fastapi import Depends, Query 

4 

5from polar.exceptions import ResourceNotFound 

6from polar.kit.pagination import ListResource, PaginationParamsQuery 

7from polar.openapi import APITag 

8from polar.postgres import AsyncReadSession, get_db_read_session 

9from polar.routing import APIRouter 

10 

11from . import auth, sorting 

12from .schemas import Member 

13from .service import member_service 

14 

15router = APIRouter( 

16 prefix="/members", 

17 tags=["members", APITag.public, APITag.mcp], 

18) 

19 

20MemberNotFound = { 

21 "description": "Member not found.", 

22 "model": ResourceNotFound.schema(), 

23} 

24 

25 

26@router.get( 

27 "/", 

28 summary="List Members", 

29 response_model=ListResource[Member], 

30) 

31async def list_members( 

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") 

45 

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 ) 

53 

54 return ListResource.from_paginated_results( 

55 [Member.model_validate(member) for member in results], 

56 count, 

57 pagination, 

58 )