Coverage for polar/customer_portal/endpoints/downloadables.py: 90%

1from fastapi import Depends, Query 

2from fastapi.responses import RedirectResponse 

3 

4from polar.benefit.schemas import BenefitID 

5from polar.kit.pagination import ListResource, PaginationParamsQuery 

6from polar.kit.schemas import MultipleQueryFilter 

7from polar.openapi import APITag 

8from polar.postgres import AsyncSession, get_db_session 

9from polar.routing import APIRouter 

10 

11from .. import auth 

12from ..schemas.downloadables import DownloadableRead 

13from ..service.downloadables import downloadable as downloadable_service 

14 

15router = APIRouter(prefix="/downloadables", tags=["downloadables", APITag.public]) 

16 

17 

18@router.get( 

19 "/", 

20 summary="List Downloadables", 

21 response_model=ListResource[DownloadableRead], 

22) 

23async def list( 

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 ) 

37 

38 return ListResource.from_paginated_results( 

39 downloadable_service.generate_downloadable_schemas(results), 

40 count, 

41 pagination, 

42 ) 

43 

44 

45@router.get( 

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( 

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)