Coverage for polar/organization_access_token/endpoints.py: 60%
36 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from fastapi import Depends, Query 1a
2from pydantic import UUID4 1a
4from polar.auth.dependencies import WebUserRead, WebUserWrite 1a
5from polar.exceptions import ResourceNotFound 1a
6from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
7from polar.kit.schemas import MultipleQueryFilter 1a
8from polar.models import OrganizationAccessToken 1a
9from polar.openapi import APITag 1a
10from polar.organization.schemas import OrganizationID 1a
11from polar.postgres import AsyncSession, get_db_session 1a
12from polar.routing import APIRouter 1a
14from . import sorting 1a
15from .schemas import ( 1a
16 OrganizationAccessToken as OrganizationAccessTokenSchema,
17)
18from .schemas import ( 1a
19 OrganizationAccessTokenCreate,
20 OrganizationAccessTokenCreateResponse,
21 OrganizationAccessTokenUpdate,
22)
23from .service import organization_access_token as organization_access_token_service 1a
25router = APIRouter( 1a
26 prefix="/organization-access-tokens",
27 tags=["organization_access_token", APITag.private],
28)
31@router.get("/", response_model=ListResource[OrganizationAccessTokenSchema]) 1a
32async def list( 1a
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 )
50 return ListResource.from_paginated_results(
51 [OrganizationAccessTokenSchema.model_validate(result) for result in results],
52 count,
53 pagination,
54 )
57@router.post("/", response_model=OrganizationAccessTokenCreateResponse, status_code=201) 1a
58async def create( 1a
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 )
74@router.patch("/{id}", response_model=OrganizationAccessTokenSchema) 1a
75async def update( 1a
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()
87 return await organization_access_token_service.update(
88 session, organization_access_token, organization_access_token_update
89 )
92@router.delete("/{id}", status_code=204) 1a
93async def delete( 1a
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()
104 await organization_access_token_service.delete(session, organization_access_token)