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

1from fastapi import Depends, Query 

2from pydantic import UUID4 

3 

4from polar.auth.dependencies import WebUserRead, WebUserWrite 

5from polar.exceptions import ResourceNotFound 

6from polar.kit.pagination import ListResource, PaginationParamsQuery 

7from polar.kit.schemas import MultipleQueryFilter 

8from polar.models import OrganizationAccessToken 

9from polar.openapi import APITag 

10from polar.organization.schemas import OrganizationID 

11from polar.postgres import AsyncSession, get_db_session 

12from polar.routing import APIRouter 

13 

14from . import sorting 

15from .schemas import ( 

16 OrganizationAccessToken as OrganizationAccessTokenSchema, 

17) 

18from .schemas import ( 

19 OrganizationAccessTokenCreate, 

20 OrganizationAccessTokenCreateResponse, 

21 OrganizationAccessTokenUpdate, 

22) 

23from .service import organization_access_token as organization_access_token_service 

24 

25router = APIRouter( 

26 prefix="/organization-access-tokens", 

27 tags=["organization_access_token", APITag.private], 

28) 

29 

30 

31@router.get("/", response_model=ListResource[OrganizationAccessTokenSchema]) 

32async def list( 

33 auth_subject: WebUserRead, 

34 pagination: PaginationParamsQuery, 

35 sorting: sorting.ListSorting, 

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

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

38 ), 

39 session: AsyncSession = Depends(get_db_session), 

40) -> ListResource[OrganizationAccessTokenSchema]: 

41 """List organization access tokens.""" 

42 results, count = await organization_access_token_service.list( 

43 session, 

44 auth_subject, 

45 organization_id=organization_id, 

46 pagination=pagination, 

47 sorting=sorting, 

48 ) 

49 

50 return ListResource.from_paginated_results( 

51 [OrganizationAccessTokenSchema.model_validate(result) for result in results], 

52 count, 

53 pagination, 

54 ) 

55 

56 

57@router.post("/", response_model=OrganizationAccessTokenCreateResponse, status_code=201) 

58async def create( 

59 organization_access_token_create: OrganizationAccessTokenCreate, 

60 auth_subject: WebUserWrite, 

61 session: AsyncSession = Depends(get_db_session), 

62) -> OrganizationAccessTokenCreateResponse: 

63 organization_access_token, token = await organization_access_token_service.create( 

64 session, auth_subject, organization_access_token_create 

65 ) 

66 return OrganizationAccessTokenCreateResponse.model_validate( 

67 { 

68 "organization_access_token": organization_access_token, 

69 "token": token, 

70 } 

71 ) 

72 

73 

74@router.patch("/{id}", response_model=OrganizationAccessTokenSchema) 

75async def update( 

76 id: UUID4, 

77 organization_access_token_update: OrganizationAccessTokenUpdate, 

78 auth_subject: WebUserWrite, 

79 session: AsyncSession = Depends(get_db_session), 

80) -> OrganizationAccessToken: 

81 organization_access_token = await organization_access_token_service.get( 

82 session, auth_subject, id 

83 ) 

84 if organization_access_token is None: 

85 raise ResourceNotFound() 

86 

87 return await organization_access_token_service.update( 

88 session, organization_access_token, organization_access_token_update 

89 ) 

90 

91 

92@router.delete("/{id}", status_code=204) 

93async def delete( 

94 id: UUID4, 

95 auth_subject: WebUserWrite, 

96 session: AsyncSession = Depends(get_db_session), 

97) -> None: 

98 organization_access_token = await organization_access_token_service.get( 

99 session, auth_subject, id 

100 ) 

101 if organization_access_token is None: 

102 raise ResourceNotFound() 

103 

104 await organization_access_token_service.delete(session, organization_access_token)