Coverage for polar/storefront/endpoints.py: 66%

1from fastapi import Depends 

2 

3from polar.exceptions import ResourceNotFound 

4from polar.kit.pagination import PaginationParams 

5from polar.models import Product 

6from polar.openapi import APITag 

7from polar.postgres import AsyncSession, get_db_session 

8from polar.routing import APIRouter 

9 

10from .schemas import OrganizationSlugLookup, Storefront 

11from .service import storefront as storefront_service 

12 

13router = APIRouter(prefix="/storefronts", tags=["storefronts", APITag.private]) 

14 

15OrganizationNotFound = { 

16 "description": "Organization not found.", 

17 "model": ResourceNotFound.schema(), 

18} 

19 

20 

21@router.get( 

22 "/{slug}", 

23 summary="Get Organization Storefront", 

24 response_model=Storefront, 

25 responses={404: OrganizationNotFound}, 

26) 

27async def get(slug: str, session: AsyncSession = Depends(get_db_session)) -> Storefront: 

28 """Get an organization storefront by slug.""" 

29 organization = await storefront_service.get(session, slug) 

30 if organization is None: 30 ↛ 34line 30 didn't jump to line 34 because the condition on line 30 was always true

31 raise ResourceNotFound() 

32 

33 # Retrieve the product that was created from the migrated donation feature 

34 donation_product: Product | None = None 

35 for product in organization.products: 

36 if product.user_metadata.get("donation_product", False): 

37 donation_product = product 

38 

39 customers, total = await storefront_service.list_customers( 

40 session, organization, pagination=PaginationParams(1, 3) 

41 ) 

42 

43 return Storefront.model_validate( 

44 { 

45 "organization": organization, 

46 "products": organization.products, 

47 "donation_product": donation_product, 

48 "customers": { 

49 "total": total, 

50 "customers": [ 

51 { 

52 "name": customer.name[0] 

53 if customer.name 

54 else customer.email[0], 

55 } 

56 for customer in customers 

57 ], 

58 }, 

59 } 

60 ) 

61 

62 

63@router.get( 

64 "/lookup/product/{product_id}", 

65 responses={404: OrganizationNotFound}, 

66) 

67async def get_organization_slug_by_product_id( 

68 product_id: str, session: AsyncSession = Depends(get_db_session) 

69) -> OrganizationSlugLookup: 

70 """Get organization slug by product ID for legacy redirect purposes.""" 

71 organization_slug = await storefront_service.get_organization_slug_by_product_id( 

72 session, product_id 

73 ) 

74 if organization_slug is None: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 raise ResourceNotFound() 

76 

77 return OrganizationSlugLookup(organization_slug=organization_slug) 

78 

79 

80@router.get( 

81 "/lookup/subscription/{subscription_id}", 

82 responses={404: OrganizationNotFound}, 

83) 

84async def get_organization_slug_by_subscription_id( 

85 subscription_id: str, session: AsyncSession = Depends(get_db_session) 

86) -> OrganizationSlugLookup: 

87 """Get organization slug by subscription ID for legacy redirect purposes.""" 

88 organization_slug = ( 

89 await storefront_service.get_organization_slug_by_subscription_id( 

90 session, subscription_id 

91 ) 

92 ) 

93 if organization_slug is None: 93 ↛ 96line 93 didn't jump to line 96 because the condition on line 93 was always true

94 raise ResourceNotFound() 

95 

96 return OrganizationSlugLookup(organization_slug=organization_slug)