Coverage for polar/storefront/endpoints.py: 39%
34 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from fastapi import Depends 1a
3from polar.exceptions import ResourceNotFound 1a
4from polar.kit.pagination import PaginationParams 1a
5from polar.models import Product 1a
6from polar.openapi import APITag 1a
7from polar.postgres import AsyncSession, get_db_session 1a
8from polar.routing import APIRouter 1a
10from .schemas import OrganizationSlugLookup, Storefront 1a
11from .service import storefront as storefront_service 1a
13router = APIRouter(prefix="/storefronts", tags=["storefronts", APITag.private]) 1a
15OrganizationNotFound = { 1a
16 "description": "Organization not found.",
17 "model": ResourceNotFound.schema(),
18}
21@router.get( 1a
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: 1ab
28 """Get an organization storefront by slug."""
29 organization = await storefront_service.get(session, slug)
30 if organization is None:
31 raise ResourceNotFound()
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
39 customers, total = await storefront_service.list_customers(
40 session, organization, pagination=PaginationParams(1, 3)
41 )
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 )
63@router.get( 1a
64 "/lookup/product/{product_id}",
65 responses={404: OrganizationNotFound},
66)
67async def get_organization_slug_by_product_id( 1ab
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:
75 raise ResourceNotFound()
77 return OrganizationSlugLookup(organization_slug=organization_slug)
80@router.get( 1a
81 "/lookup/subscription/{subscription_id}",
82 responses={404: OrganizationNotFound},
83)
84async def get_organization_slug_by_subscription_id( 1ab
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:
94 raise ResourceNotFound()
96 return OrganizationSlugLookup(organization_slug=organization_slug)