Coverage for polar/models/oauth2_token.py: 81%

1from typing import TYPE_CHECKING, Any, cast 

2 

3from authlib.integrations.sqla_oauth2 import OAuth2TokenMixin 

4from sqlalchemy import String 

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

6 

7from polar.auth.scope import Scope, scope_to_set 

8from polar.kit.db.models import RecordModel 

9from polar.oauth2.sub_type import SubTypeModelMixin 

10 

11if TYPE_CHECKING: 11 ↛ 12line 11 didn't jump to line 12 because the condition on line 11 was never true

12 from .oauth2_client import OAuth2Client 

13 

14 

15class OAuth2Token(RecordModel, OAuth2TokenMixin, SubTypeModelMixin): 

16 __tablename__ = "oauth2_tokens" 

17 

18 client_id: Mapped[str] = mapped_column(String(52), nullable=False) 

19 nonce: Mapped[str | None] = mapped_column(String, index=True, nullable=True) 

20 

21 @declared_attr 

22 def client(cls) -> "Mapped[OAuth2Client]": 

23 return relationship( 

24 "OAuth2Client", 

25 primaryjoin="foreign(OAuth2Token.client_id) == OAuth2Client.client_id", 

26 viewonly=True, 

27 lazy="raise", 

28 ) 

29 

30 @property 

31 def expires_at(self) -> int: 

32 return cast(int, self.issued_at) + cast(int, self.expires_in) 

33 

34 @property 

35 def scopes(self) -> set[Scope]: 

36 return scope_to_set(cast(str, self.get_scope())) 

37 

38 def get_introspection_data(self, issuer: str) -> dict[str, Any]: 

39 return { 

40 "active": not cast(bool, self.is_revoked()) 

41 and not cast(bool, self.is_expired()), 

42 "client_id": self.client_id, 

43 "token_type": self.token_type, 

44 "scope": self.get_scope(), 

45 "sub_type": self.sub_type, 

46 "sub": str(self.sub.id), 

47 "aud": self.client_id, 

48 "iss": issuer, 

49 "exp": self.expires_at, 

50 "iat": self.issued_at, 

51 }