Coverage for polar/customer_portal/endpoints/downloadables.py: 76%
21 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from fastapi import Depends, Query 1a
2from fastapi.responses import RedirectResponse 1a
4from polar.benefit.schemas import BenefitID 1a
5from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
6from polar.kit.schemas import MultipleQueryFilter 1a
7from polar.openapi import APITag 1a
8from polar.postgres import AsyncSession, get_db_session 1a
9from polar.routing import APIRouter 1a
11from .. import auth 1a
12from ..schemas.downloadables import DownloadableRead 1a
13from ..service.downloadables import downloadable as downloadable_service 1a
15router = APIRouter(prefix="/downloadables", tags=["downloadables", APITag.public]) 1a
18@router.get( 1a
19 "/",
20 summary="List Downloadables",
21 response_model=ListResource[DownloadableRead],
22)
23async def list( 1a
24 auth_subject: auth.CustomerPortalRead,
25 pagination: PaginationParamsQuery,
26 benefit_id: MultipleQueryFilter[BenefitID] | None = Query(
27 None, title="BenefitID Filter", description="Filter by benefit ID."
28 ),
29 session: AsyncSession = Depends(get_db_session),
30) -> ListResource[DownloadableRead]:
31 results, count = await downloadable_service.get_list(
32 session,
33 auth_subject,
34 pagination=pagination,
35 benefit_id=benefit_id,
36 )
38 return ListResource.from_paginated_results(
39 downloadable_service.generate_downloadable_schemas(results),
40 count,
41 pagination,
42 )
45@router.get( 1a
46 "/{token}",
47 summary="Get Downloadable",
48 responses={
49 302: {"description": "Redirected to download"},
50 400: {"description": "Invalid signature"},
51 404: {"description": "Downloadable not found"},
52 410: {"description": "Expired signature"},
53 },
54 name="customer_portal.downloadables.get",
55 tags=[APITag.private],
56)
57async def get( 1ab
58 token: str,
59 session: AsyncSession = Depends(get_db_session),
60) -> RedirectResponse:
61 downloadable = await downloadable_service.get_from_token_or_raise(
62 session, token=token
63 )
64 signed = downloadable_service.generate_download_schema(downloadable)
65 return RedirectResponse(signed.file.download.url, 302)