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

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

2 

3from abc import abstractmethod 

4from typing import AsyncIterator 

5 

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

7 RefreshTokenEntity, 

8 RefreshTokenIdentifier, 

9) 

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

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

12 

13 

14class RefreshTokenNotFoundException(Exception): 

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

16 

17 

18class RefreshTokenRepository: 

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

20 

21 @abstractmethod 

22 def create_query(self) -> RefreshTokenQuery: 

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

24 raise NotImplementedError 

25 

26 @abstractmethod 

27 async def get_by_token_identifier( 

28 self, identifier: TokenIdentifier 

29 ) -> RefreshTokenEntity: 

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

31 raise NotImplementedError 

32 

33 @abstractmethod 

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

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

36 raise NotImplementedError 

37 

38 @abstractmethod 

39 async def get_all( 

40 self, 

41 query: RefreshTokenQuery | None = None, 

42 limit: int | None = None, 

43 offset: int | None = None, 

44 ) -> AsyncIterator[RefreshTokenEntity]: 

45 """Get all refresh tokens.""" 

46 raise NotImplementedError 

47 

48 @abstractmethod 

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

50 """Save a new refresh token.""" 

51 raise NotImplementedError 

52 

53 @abstractmethod 

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

55 """Update the refresh token.""" 

56 raise NotImplementedError