Coverage for polar/oauth2/service/oauth2_authorization_code.py: 54%

22 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1import structlog 1a

2from sqlalchemy import select 1a

3 

4from polar.config import settings 1a

5from polar.enums import TokenType 1a

6from polar.kit.crypto import get_token_hash 1a

7from polar.kit.services import ResourceServiceReader 1a

8from polar.logging import Logger 1a

9from polar.models import OAuth2AuthorizationCode 1a

10from polar.postgres import AsyncSession 1a

11 

12log: Logger = structlog.get_logger() 1a

13 

14 

15class OAuth2AuthorizationCodeService(ResourceServiceReader[OAuth2AuthorizationCode]): 1a

16 async def revoke_leaked( 1a

17 self, 

18 session: AsyncSession, 

19 token: str, 

20 token_type: TokenType, 

21 *, 

22 notifier: str, 

23 url: str | None = None, 

24 ) -> bool: 

25 statement = select(OAuth2AuthorizationCode).where( 

26 OAuth2AuthorizationCode.code 

27 == get_token_hash(token, secret=settings.SECRET) 

28 ) 

29 

30 result = await session.execute(statement) 

31 authorization_code = result.unique().scalar_one_or_none() 

32 

33 if authorization_code is None: 

34 return False 

35 

36 authorization_code.set_deleted_at() 

37 session.add(authorization_code) 

38 

39 log.info( 

40 "Revoke leaked authorization code", 

41 id=authorization_code.id, 

42 notifier=notifier, 

43 url=url, 

44 ) 

45 

46 return True 

47 

48 

49oauth2_authorization_code = OAuth2AuthorizationCodeService(OAuth2AuthorizationCode) 1a