Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/tokens/access_token.py: 95%

21 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2024-01-01 00:00 +0000

1"""Module that defines an access token entity.""" 

2 

3from dataclasses import dataclass, field, replace 

4from typing import ClassVar, Self, Type 

5 

6from kwai_core.domain.entity import DataclassEntity 

7from kwai_core.domain.value_objects.identifier import IntIdentifier 

8from kwai_core.domain.value_objects.timestamp import Timestamp 

9 

10from kwai_bc_identity.tokens.token_identifier import TokenIdentifier 

11from kwai_bc_identity.users.user_account import UserAccountEntity 

12 

13 

14class AccessTokenIdentifier(IntIdentifier): 

15 """Identifier for an access token.""" 

16 

17 

18@dataclass(kw_only=True, eq=False, slots=True, frozen=True) 

19class AccessTokenEntity(DataclassEntity): 

20 """An access token entity. 

21 

22 Attributes: 

23 user_account: The user account associated with this token. 

24 identifier: The actual token. 

25 expiration: The expiration timestamp of the token. 

26 revoked: Whether the token has been revoked. 

27 """ 

28 

29 ID: ClassVar[Type] = AccessTokenIdentifier 

30 

31 user_account: UserAccountEntity 

32 identifier: TokenIdentifier = field(default_factory=TokenIdentifier.generate) 

33 expiration: Timestamp = field(default_factory=Timestamp.create_now) 

34 revoked: bool = field(default=False) 

35 

36 @property 

37 def expired(self) -> bool: 

38 """Return true when the access token is expired.""" 

39 return self.expiration.is_past 

40 

41 def revoke(self) -> Self: 

42 """Revoke the access token. 

43 

44 Returns: 

45 A revoked access token. 

46 """ 

47 return replace( 

48 self, revoked=True, traceable_time=self.traceable_time.mark_for_update() 

49 ) 

50 

51 def renew(self, expiry_minutes: int) -> Self: 

52 """Renew the access token.""" 

53 return replace( 

54 self, 

55 identifier=TokenIdentifier.generate(), 

56 expiration=Timestamp.create_with_delta(minutes=expiry_minutes), 

57 traceable_time=self.traceable_time.mark_for_update(), 

58 )