Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/tokens/refresh_token.py: 100%

1"""Module for a refresh 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.access_token import AccessTokenEntity 

11from kwai_bc_identity.tokens.token_identifier import TokenIdentifier 

12 

13 

14class RefreshTokenIdentifier(IntIdentifier): 

15 """Identifier for a refresh token.""" 

16 

17 

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

19class RefreshTokenEntity(DataclassEntity): 

20 """A refresh token entity. 

21 

22 Attributes: 

23 access_token: The access token associated with this refresh 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] = RefreshTokenIdentifier 

30 

31 access_token: AccessTokenEntity 

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 token is expired.""" 

39 return self.expiration.is_past 

40 

41 def revoke(self) -> Self: 

42 """Revoke the refresh token. 

43 

44 Returns: 

45 A revoked refresh 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, access_token_expiry_minutes: int) -> Self: 

52 """Renew the refresh token. 

53 

54 The access token will also be renewed. 

55 

56 Args: 

57 expiry_minutes: The number of minutes until the refresh token expires. 

58 access_token_expiry_minutes: 

59 The number of minutes until the access token expires. 

60 """ 

61 return replace( 

62 self, 

63 identifier=TokenIdentifier.generate(), 

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

65 access_token=self.access_token.renew(access_token_expiry_minutes), 

66 traceable_time=self.traceable_time.mark_for_update(), 

67 )