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

1from datetime import datetime 2 ctx1ab

2from urllib.parse import urlencode 2 ctx1ab

3from uuid import UUID 2 ctx1ab

4 

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

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

7 

8from polar.config import settings 2 ctx1ab

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

10from polar.kit.utils import utc_now 2 ctx1ab

11 

12from .customer import Customer 2 ctx1ab

13 

14 

15def get_expires_at() -> datetime: 2 ctx1ab

16 return utc_now() + settings.CUSTOMER_SESSION_TTL 

17 

18 

19class CustomerSession(RecordModel): 2 ctx1ab

20 __tablename__ = "customer_sessions" 2 ctx1ab

21 

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

23 expires_at: Mapped[datetime] = mapped_column( 2 ctx1ab

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) 2 ctx1ab

27 

28 customer_id: Mapped[UUID] = mapped_column( 2 ctx1ab

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

30 ) 

31 

32 @declared_attr 2 ctx1ab

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

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

35 

36 @property 2 ctx1ab

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

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

39 

40 @raw_token.setter 2 ctx1ab

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

42 self._raw_token = value 

43 

44 @property 2 ctx1ab

45 def customer_portal_url(self) -> str: 2 ctx1ab

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 )