Coverage for polar/checkout/repository.py: 36%
35 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from uuid import UUID 1a
3from sqlalchemy import Select, select, update 1a
4from sqlalchemy.orm import joinedload, selectinload 1a
6from polar.auth.models import AuthSubject, User, is_organization, is_user 1a
7from polar.kit.repository import ( 1a
8 Options,
9 RepositoryBase,
10 RepositorySoftDeletionIDMixin,
11 RepositorySoftDeletionMixin,
12 RepositorySortingMixin,
13 SortingClause,
14)
15from polar.kit.utils import utc_now 1a
16from polar.models import ( 1a
17 Checkout,
18 CheckoutProduct,
19 Organization,
20 Product,
21 UserOrganization,
22)
23from polar.models.checkout import CheckoutStatus 1a
25from .sorting import CheckoutSortProperty 1a
28class CheckoutRepository( 1a
29 RepositorySortingMixin[Checkout, CheckoutSortProperty],
30 RepositorySoftDeletionIDMixin[Checkout, UUID],
31 RepositorySoftDeletionMixin[Checkout],
32 RepositoryBase[Checkout],
33):
34 model = Checkout 1a
36 async def get_by_client_secret( 1a
37 self, client_secret: str, *, options: Options = ()
38 ) -> Checkout | None:
39 statement = (
40 self.get_base_statement()
41 .where(Checkout.client_secret == client_secret)
42 .options(*options)
43 )
44 return await self.get_one_or_none(statement)
46 async def expire_open_checkouts(self) -> None: 1a
47 statement = (
48 update(Checkout)
49 .where(
50 Checkout.deleted_at.is_(None),
51 Checkout.expires_at <= utc_now(),
52 Checkout.status == CheckoutStatus.open,
53 )
54 .values(status=CheckoutStatus.expired)
55 )
56 await self.session.execute(statement)
58 def get_readable_statement( 1a
59 self, auth_subject: AuthSubject[User | Organization]
60 ) -> Select[tuple[Checkout]]:
61 statement = self.get_base_statement()
63 if is_user(auth_subject):
64 user = auth_subject.subject
65 statement = statement.where(
66 Checkout.organization_id.in_(
67 select(UserOrganization.organization_id).where(
68 UserOrganization.user_id == user.id,
69 UserOrganization.deleted_at.is_(None),
70 )
71 )
72 )
73 elif is_organization(auth_subject):
74 statement = statement.where(
75 Checkout.organization_id == auth_subject.subject.id,
76 )
78 return statement
80 def get_eager_options(self) -> Options: 1a
81 return (
82 joinedload(Checkout.organization).joinedload(Organization.account),
83 joinedload(Checkout.customer),
84 joinedload(Checkout.product).options(
85 selectinload(Product.product_medias),
86 selectinload(Product.attached_custom_fields),
87 ),
88 selectinload(Checkout.checkout_products).options(
89 joinedload(CheckoutProduct.product).options(
90 selectinload(Product.product_medias),
91 )
92 ),
93 joinedload(Checkout.subscription),
94 joinedload(Checkout.discount),
95 joinedload(Checkout.product_price),
96 )
98 def get_sorting_clause(self, property: CheckoutSortProperty) -> SortingClause: 1a
99 match property:
100 case CheckoutSortProperty.created_at:
101 return Checkout.created_at
102 case CheckoutSortProperty.expires_at:
103 return Checkout.expires_at
104 case CheckoutSortProperty.status:
105 return Checkout.status