Coverage for polar/benefit/grant/endpoints.py: 74%

1from fastapi import Depends, Query 

2 

3from polar.customer.schemas.customer import CustomerID 

4from polar.kit.pagination import ListResource, PaginationParamsQuery 

5from polar.kit.schemas import MultipleQueryFilter 

6from polar.openapi import APITag 

7from polar.organization.schemas import OrganizationID 

8from polar.postgres import AsyncSession, get_db_session 

9from polar.routing import APIRouter 

10 

11from ..auth import BenefitsRead 

12from ..schemas import BenefitGrant 

13from .service import benefit_grant as benefit_grant_service 

14from .sorting import ListSorting 

15 

16router = APIRouter(prefix="/benefit-grants", tags=["benefit-grants", APITag.public]) 

17 

18 

19@router.get( 

20 "/", 

21 response_model=ListResource[BenefitGrant], 

22 summary="List Benefit Grants", 

23) 

24async def list( 

25 auth_subject: BenefitsRead, 

26 pagination: PaginationParamsQuery, 

27 sorting: ListSorting, 

28 organization_id: MultipleQueryFilter[OrganizationID] | None = Query( 

29 None, title="OrganizationID Filter", description="Filter by organization ID." 

30 ), 

31 customer_id: MultipleQueryFilter[CustomerID] | None = Query( 

32 None, title="CustomerID Filter", description="Filter by customer ID." 

33 ), 

34 is_granted: bool | None = Query( 

35 None, 

36 description=( 

37 "Filter by granted status. " 

38 "If `true`, only granted benefits will be returned. " 

39 "If `false`, only revoked benefits will be returned. " 

40 ), 

41 ), 

42 session: AsyncSession = Depends(get_db_session), 

43) -> ListResource[BenefitGrant]: 

44 """List benefit grants across all benefits for the authenticated organization.""" 

45 

46 # Extract the first organization_id if provided, otherwise use the auth subject's organization 

47 if organization_id is not None and len(organization_id) > 0: 47 ↛ 51line 47 didn't jump to line 51 because the condition on line 47 was always true

48 org_id = organization_id[0] 

49 else: 

50 # Use the authenticated organization 

51 if hasattr(auth_subject.subject, "id"): 

52 org_id = auth_subject.subject.id 

53 else: 

54 # If auth subject doesn't have organization, we need to handle this case 

55 # For now, require organization_id to be provided 

56 from polar.exceptions import BadRequest 

57 

58 raise BadRequest("organization_id parameter is required") 

59 

60 results, count = await benefit_grant_service.list_by_organization( 

61 session, 

62 org_id, 

63 is_granted=is_granted, 

64 customer_id=customer_id, 

65 pagination=pagination, 

66 sorting=sorting, 

67 ) 

68 

69 return ListResource.from_paginated_results( 

70 [BenefitGrant.model_validate(result) for result in results], 

71 count, 

72 pagination, 

73 )