Coverage for kwai/modules/identity/tokens/refresh_token_repository.py: 100%

1"""Module that defines an interface for a refresh token repository.""" 

2from abc import abstractmethod 

3from typing import AsyncIterator 

4 

5from kwai.modules.identity.tokens.refresh_token import ( 

6 RefreshTokenEntity, 

7 RefreshTokenIdentifier, 

8) 

9from kwai.modules.identity.tokens.refresh_token_query import RefreshTokenQuery 

10from kwai.modules.identity.tokens.token_identifier import TokenIdentifier 

11 

12 

13class RefreshTokenNotFoundException(Exception): 

14 """Raised when the refresh token can't be found.""" 

15 

16 

17class RefreshTokenRepository: 

18 """Interface for a refresh token repository.""" 

19 

20 @abstractmethod 

21 def create_query(self) -> RefreshTokenQuery: 

22 """Create a query for a refresh token.""" 

23 raise NotImplementedError 

24 

25 @abstractmethod 

26 async def get_by_token_identifier( 

27 self, identifier: TokenIdentifier 

28 ) -> RefreshTokenEntity: 

29 """Get the refresh token with the given token identifier.""" 

30 raise NotImplementedError 

31 

32 @abstractmethod 

33 async def get(self, id_: RefreshTokenIdentifier) -> RefreshTokenEntity: 

34 """Get the refresh token entity with the given id.""" 

35 raise NotImplementedError 

36 

37 @abstractmethod 

38 async def get_all( 

39 self, 

40 query: RefreshTokenQuery | None = None, 

41 limit: int | None = None, 

42 offset: int | None = None, 

43 ) -> AsyncIterator[RefreshTokenEntity]: 

44 """Get all refresh tokens.""" 

45 raise NotImplementedError 

46 

47 @abstractmethod 

48 async def create(self, refresh_token: RefreshTokenEntity) -> RefreshTokenEntity: 

49 """Save a new refresh token.""" 

50 raise NotImplementedError 

51 

52 @abstractmethod 

53 async def update(self, refresh_token: RefreshTokenEntity): 

54 """Update the refresh token.""" 

55 raise NotImplementedError