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

1"""Module that defines an interface for an access token repository.""" 

2from abc import abstractmethod 

3from typing import Iterator 

4 

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

6 AccessTokenEntity, 

7 AccessTokenIdentifier, 

8) 

9from kwai.modules.identity.tokens.access_token_query import AccessTokenQuery 

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

11 

12 

13class AccessTokenNotFoundException(Exception): 

14 """Raised when the access token could not be found.""" 

15 

16 

17class AccessTokenRepository: 

18 """Interface for an access token repository.""" 

19 

20 @abstractmethod 

21 def create_query(self) -> AccessTokenQuery: 

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

23 raise NotImplementedError 

24 

25 @abstractmethod 

26 async def create(self, access_token: AccessTokenEntity) -> AccessTokenEntity: 

27 """Save a new access token.""" 

28 raise NotImplementedError 

29 

30 @abstractmethod 

31 async def update(self, access_token: AccessTokenEntity): 

32 """Update the access token.""" 

33 raise NotImplementedError 

34 

35 @abstractmethod 

36 async def get(self, id_: AccessTokenIdentifier) -> AccessTokenEntity: 

37 """Get the access token with the given id.""" 

38 raise NotImplementedError 

39 

40 @abstractmethod 

41 async def get_by_identifier(self, identifier: TokenIdentifier) -> AccessTokenEntity: 

42 """Get the access token with the given identifier.""" 

43 raise NotImplementedError 

44 

45 @abstractmethod 

46 async def get_all( 

47 self, 

48 query: AccessTokenQuery | None = None, 

49 limit: int | None = None, 

50 offset: int | None = None, 

51 ) -> Iterator[AccessTokenEntity]: 

52 """Get all access token.""" 

53 raise NotImplementedError