Coverage for polar/models/customer_session.py: 83%

1from datetime import datetime 

2from urllib.parse import urlencode 

3from uuid import UUID 

4 

5from sqlalchemy import CHAR, TIMESTAMP, ForeignKey, String, Uuid 

6from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship 

7 

8from polar.config import settings 

9from polar.kit.db.models.base import RecordModel 

10from polar.kit.utils import utc_now 

11 

12from .customer import Customer 

13 

14 

15def get_expires_at() -> datetime: 

16 return utc_now() + settings.CUSTOMER_SESSION_TTL 

17 

18 

19class CustomerSession(RecordModel): 

20 __tablename__ = "customer_sessions" 

21 

22 token: Mapped[str] = mapped_column(CHAR(64), unique=True, nullable=False) 

23 expires_at: Mapped[datetime] = mapped_column( 

24 TIMESTAMP(timezone=True), nullable=False, index=True, default=get_expires_at 

25 ) 

26 return_url: Mapped[str | None] = mapped_column(String, nullable=True, default=None) 

27 

28 customer_id: Mapped[UUID] = mapped_column( 

29 Uuid, ForeignKey("customers.id", ondelete="cascade"), nullable=False, index=True 

30 ) 

31 

32 @declared_attr 

33 def customer(cls) -> Mapped[Customer]: 

34 return relationship(Customer, lazy="joined") 

35 

36 @property 

37 def raw_token(self) -> str | None: 

38 return getattr(self, "_raw_token", None) 

39 

40 @raw_token.setter 

41 def raw_token(self, value: str) -> None: 

42 self._raw_token = value 

43 

44 @property 

45 def customer_portal_url(self) -> str: 

46 query_string = urlencode( 

47 {"customer_session_token": self.raw_token, "email": self.customer.email} 

48 ) 

49 return settings.generate_frontend_url( 

50 f"/{self.customer.organization.slug}/portal?{query_string}" 

51 )