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

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

2 

3from abc import abstractmethod 

4from typing import Iterator 

5 

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

7 AccessTokenEntity, 

8 AccessTokenIdentifier, 

9) 

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

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

12 

13 

14class AccessTokenNotFoundException(Exception): 

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

16 

17 

18class AccessTokenRepository: 

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

20 

21 @abstractmethod 

22 def create_query(self) -> AccessTokenQuery: 

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

24 raise NotImplementedError 

25 

26 @abstractmethod 

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

28 """Save a new access token.""" 

29 raise NotImplementedError 

30 

31 @abstractmethod 

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

33 """Update the access token.""" 

34 raise NotImplementedError 

35 

36 @abstractmethod 

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

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

39 raise NotImplementedError 

40 

41 @abstractmethod 

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

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

44 raise NotImplementedError 

45 

46 @abstractmethod 

47 async def get_all( 

48 self, 

49 query: AccessTokenQuery | None = None, 

50 limit: int | None = None, 

51 offset: int | None = None, 

52 ) -> Iterator[AccessTokenEntity]: 

53 """Get all access token.""" 

54 raise NotImplementedError