Coverage for polar/personal_access_token/endpoints.py: 74%

1from fastapi import Depends 

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.openapi import APITag 

8from polar.postgres import AsyncSession, get_db_session 

9from polar.routing import APIRouter 

10 

11from .schemas import PersonalAccessToken 

12from .service import personal_access_token as personal_access_token_service 

13 

14router = APIRouter( 

15 prefix="/personal_access_tokens", tags=["personal_access_token", APITag.private] 

16) 

17 

18 

19@router.get("/", response_model=ListResource[PersonalAccessToken]) 

20async def list_personal_access_tokens( 

21 auth_subject: WebUserRead, 

22 pagination: PaginationParamsQuery, 

23 session: AsyncSession = Depends(get_db_session), 

24) -> ListResource[PersonalAccessToken]: 

25 """List personal access tokens.""" 

26 results, count = await personal_access_token_service.list( 

27 session, auth_subject, pagination=pagination 

28 ) 

29 

30 return ListResource.from_paginated_results( 

31 [PersonalAccessToken.model_validate(result) for result in results], 

32 count, 

33 pagination, 

34 ) 

35 

36 

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

38async def delete_personal_access_token( 

39 id: UUID4, 

40 auth_subject: WebUserWrite, 

41 session: AsyncSession = Depends(get_db_session), 

42) -> None: 

43 personal_access_token = await personal_access_token_service.get_by_id( 

44 session, auth_subject, id 

45 ) 

46 if personal_access_token is None: 

47 raise ResourceNotFound() 

48 

49 await personal_access_token_service.delete(session, personal_access_token)