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

21 statements  

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

1from fastapi import Depends 1a

2from pydantic import UUID4 1a

3 

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

5from polar.exceptions import ResourceNotFound 1a

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

7from polar.openapi import APITag 1a

8from polar.postgres import AsyncSession, get_db_session 1a

9from polar.routing import APIRouter 1a

10 

11from .schemas import PersonalAccessToken 1a

12from .service import personal_access_token as personal_access_token_service 1a

13 

14router = APIRouter( 1a

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

16) 

17 

18 

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

20async def list_personal_access_tokens( 1a

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) 1a

38async def delete_personal_access_token( 1a

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)