Coverage for polar/checkout_link/repository.py: 35%
36 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import TYPE_CHECKING 1a
2from uuid import UUID 1a
4from sqlalchemy import Select, select 1a
5from sqlalchemy.orm import joinedload, selectinload 1a
7from polar.auth.models import AuthSubject, Organization, User, is_organization, is_user 1a
8from polar.kit.repository import ( 1a
9 Options,
10 RepositoryBase,
11 RepositorySoftDeletionIDMixin,
12 RepositorySoftDeletionMixin,
13)
14from polar.models import CheckoutLink, CheckoutLinkProduct, Product, UserOrganization 1a
16if TYPE_CHECKING: 16 ↛ 17line 16 didn't jump to line 17 because the condition on line 16 was never true1a
17 from sqlalchemy.orm.strategy_options import _AbstractLoad
20class CheckoutLinkRepository( 1a
21 RepositorySoftDeletionIDMixin[CheckoutLink, UUID],
22 RepositorySoftDeletionMixin[CheckoutLink],
23 RepositoryBase[CheckoutLink],
24):
25 model = CheckoutLink 1a
27 async def get_by_client_secret( 1a
28 self, client_secret: str, *, options: Options = ()
29 ) -> CheckoutLink | None:
30 statement = (
31 self.get_base_statement()
32 .join(
33 Organization, onclause=Organization.id == CheckoutLink.organization_id
34 )
35 .where(
36 CheckoutLink.client_secret == client_secret,
37 Organization.deleted_at.is_(None),
38 Organization.blocked_at.is_(None),
39 )
40 .options(*options)
41 )
42 return await self.get_one_or_none(statement)
44 def get_eager_options( 1a
45 self, *, checkout_link_product_load: "_AbstractLoad | None" = None
46 ) -> Options:
47 checkout_link_product_load = (
48 checkout_link_product_load
49 if checkout_link_product_load
50 else selectinload(CheckoutLink.checkout_link_products)
51 )
52 return (
53 checkout_link_product_load.options(
54 joinedload(CheckoutLinkProduct.product).options(
55 joinedload(Product.organization),
56 joinedload(Product.product_medias),
57 joinedload(Product.attached_custom_fields),
58 )
59 ),
60 joinedload(CheckoutLink.discount),
61 joinedload(CheckoutLink.organization),
62 )
64 def get_readable_statement( 1a
65 self, auth_subject: AuthSubject[User | Organization]
66 ) -> Select[tuple[CheckoutLink]]:
67 statement = self.get_base_statement()
69 if is_user(auth_subject):
70 user = auth_subject.subject
71 statement = statement.where(
72 CheckoutLink.organization_id.in_(
73 select(UserOrganization.organization_id).where(
74 UserOrganization.user_id == user.id,
75 UserOrganization.deleted_at.is_(None),
76 )
77 )
78 )
79 elif is_organization(auth_subject):
80 statement = statement.where(
81 CheckoutLink.organization_id == auth_subject.subject.id,
82 )
84 return statement
86 async def count_by_organization_id(self, organization_id: UUID) -> int: 1a
87 """Count checkout links for a specific organization."""
88 statement = self.get_base_statement().where(
89 CheckoutLink.organization_id == organization_id
90 )
91 return await self.count(statement)
93 async def archive_product(self, product_id: UUID) -> None: 1a
94 statement = (
95 self.get_base_statement()
96 .where(
97 CheckoutLink.id.in_(
98 select(CheckoutLinkProduct.checkout_link_id).where(
99 CheckoutLinkProduct.product_id == product_id
100 )
101 )
102 )
103 .options(selectinload(CheckoutLink.checkout_link_products))
104 )
105 checkout_links = await self.get_all(statement)
106 for checkout_link in checkout_links:
107 checkout_link.checkout_link_products = [
108 product
109 for product in checkout_link.checkout_link_products
110 if product.product_id != product_id
111 ]
112 if not checkout_link.checkout_link_products:
113 await self.soft_delete(checkout_link)
114 self.session.add(checkout_link)