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

1from typing import TYPE_CHECKING, Any, cast 2 ctx1ab

2 

3from authlib.integrations.sqla_oauth2 import OAuth2TokenMixin 2 ctx1ab

4from sqlalchemy import String 2 ctx1ab

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

6 

7from polar.auth.scope import Scope, scope_to_set 2 ctx1ab

8from polar.kit.db.models import RecordModel 2 ctx1ab

9from polar.oauth2.sub_type import SubTypeModelMixin 2 ctx1ab

10 

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

12 from .oauth2_client import OAuth2Client 

13 

14 

15class OAuth2Token(RecordModel, OAuth2TokenMixin, SubTypeModelMixin): 2 ctx1ab

16 __tablename__ = "oauth2_tokens" 2 ctx1ab

17 

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

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

20 

21 @declared_attr 2 ctx1ab

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

23 return relationship( 2 ctx1ab

24 "OAuth2Client", 

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

26 viewonly=True, 

27 lazy="raise", 

28 ) 

29 

30 @property 2 ctx1ab

31 def expires_at(self) -> int: 2 ctx1ab

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

33 

34 @property 2 ctx1ab

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

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

37 

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

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 }