Coverage for polar/account/endpoints.py: 36%

59 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 15:52 +0000

1from uuid import UUID 1a

2 

3from fastapi import Depends, Query 1a

4 

5from polar.auth.dependencies import WebUserRead, WebUserWrite 1a

6from polar.enums import AccountType 1a

7from polar.exceptions import InternalServerError, ResourceNotFound 1a

8from polar.kit.pagination import ListResource, PaginationParamsQuery 1a

9from polar.models import Account 1a

10from polar.openapi import APITag 1a

11from polar.organization.service import organization as organization_service 1a

12from polar.postgres import ( 1a

13 AsyncReadSession, 

14 AsyncSession, 

15 get_db_read_session, 

16 get_db_session, 

17) 

18from polar.routing import APIRouter 1a

19 

20from .schemas import Account as AccountSchema 1a

21from .schemas import AccountCreateForOrganization, AccountLink, AccountUpdate 1a

22from .service import account as account_service 1a

23 

24router = APIRouter(tags=["accounts", APITag.private]) 1a

25 

26 

27@router.get("/accounts/search", response_model=ListResource[AccountSchema]) 1a

28async def search( 1a

29 auth_subject: WebUserRead, 

30 pagination: PaginationParamsQuery, 

31 session: AsyncReadSession = Depends(get_db_read_session), 

32) -> ListResource[AccountSchema]: 

33 results, count = await account_service.search( 

34 session, auth_subject, pagination=pagination 

35 ) 

36 

37 return ListResource.from_paginated_results( 

38 [AccountSchema.model_validate(result) for result in results], 

39 count, 

40 pagination, 

41 ) 

42 

43 

44@router.get("/accounts/{id}", response_model=AccountSchema) 1a

45async def get( 1a

46 id: UUID, 

47 auth_subject: WebUserRead, 

48 session: AsyncReadSession = Depends(get_db_read_session), 

49) -> Account: 

50 account = await account_service.get(session, auth_subject, id) 

51 if account is None: 

52 raise ResourceNotFound() 

53 

54 return account 

55 

56 

57@router.post("/accounts", response_model=AccountSchema) 1a

58async def create( 1a

59 account_create: AccountCreateForOrganization, 

60 auth_subject: WebUserWrite, 

61 session: AsyncSession = Depends(get_db_session), 

62) -> Account: 

63 organization = await organization_service.get( 

64 session, auth_subject, account_create.organization_id 

65 ) 

66 if organization is None: 

67 raise ResourceNotFound("Organization not found") 

68 

69 account = await account_service.get_or_create_account_for_organization( 

70 session, 

71 organization=organization, 

72 admin=auth_subject.subject, 

73 account_create=account_create, 

74 ) 

75 

76 return account 

77 

78 

79@router.patch("/accounts/{id}", response_model=AccountSchema) 1a

80async def patch( 1a

81 id: UUID, 

82 account_update: AccountUpdate, 

83 auth_subject: WebUserWrite, 

84 session: AsyncSession = Depends(get_db_session), 

85) -> Account: 

86 account = await account_service.get(session, auth_subject, id) 

87 if account is None: 

88 raise ResourceNotFound() 

89 

90 return await account_service.update(session, account, account_update) 

91 

92 

93@router.post("/accounts/{id}/onboarding_link", response_model=AccountLink) 1a

94async def onboarding_link( 1a

95 id: UUID, 

96 auth_subject: WebUserWrite, 

97 return_path: str = Query(...), 

98 session: AsyncSession = Depends(get_db_session), 

99) -> AccountLink: 

100 account = await account_service.get(session, auth_subject, id) 

101 if account is None: 

102 raise ResourceNotFound() 

103 

104 if account.account_type != AccountType.stripe: 

105 raise ResourceNotFound() 

106 

107 link = await account_service.onboarding_link(account, return_path) 

108 if not link: 

109 raise InternalServerError("Failed to create link") 

110 

111 return link 

112 

113 

114@router.post("/accounts/{id}/dashboard_link", response_model=AccountLink) 1a

115async def dashboard_link( 1a

116 id: UUID, 

117 auth_subject: WebUserWrite, 

118 session: AsyncSession = Depends(get_db_session), 

119) -> AccountLink: 

120 account = await account_service.get(session, auth_subject, id) 

121 if account is None: 

122 raise ResourceNotFound() 

123 

124 # update stripe account details 

125 await account_service.sync_to_upstream(session, account) 

126 

127 link = await account_service.dashboard_link(account) 

128 if not link: 

129 raise InternalServerError("Failed to create link") 

130 

131 return link