Coverage for /usr/lib/python3.10/site-packages/hyd/backend/token/service.py: 84%

1import datetime as dt 

2 

3from sqlalchemy.exc import IntegrityError 

4from sqlalchemy.orm import Session 

5 

6from hyd.backend.exc import UnknownTokenError 

7from hyd.backend.token.models import TokenEntry, TokenScopeEntry 

8from hyd.backend.util.logger import HydLogger 

9from hyd.backend.util.models import PrimaryKey 

10 

11UTC = dt.timezone.utc 

12LOGGER = HydLogger("TokenService") 

13 

14 

15def create_token( 

16 *, 

17 user_id: PrimaryKey, 

18 expires_on: dt.datetime | None, 

19 scopes: list[str], 

20 is_login_token: bool, 

21 project_id: PrimaryKey | None, 

22 comment: str, 

23 db: Session, 

24) -> TokenEntry: 

25 token_entry = TokenEntry( 

26 user_id=user_id, 

27 expires_on=expires_on, 

28 is_login_token=is_login_token, 

29 project_id=project_id, 

30 comment=comment, 

31 ) 

32 db.add(token_entry) 

33 db.commit() 

34 

35 token_id = token_entry.id 

36 for scope in scopes: 

37 scope_entry = TokenScopeEntry(token_id=token_id, scope=scope) 

38 db.add(scope_entry) 

39 

40 db.commit() 

41 LOGGER.debug("TokenEntry created {id: %d, scopes: %s}", token_id, scopes) 

42 return token_entry 

43 

44 

45def read_token(*, token_id: int, db: Session) -> TokenEntry: 

46 try: 

47 token_entry = db.query(TokenEntry).get(token_id) 

48 except IntegrityError: 

49 raise UnknownTokenError 

50 return token_entry 

51 

52 

53def read_tokens(*, db: Session) -> list[TokenEntry]: 

54 return db.query(TokenEntry).all() 

55 

56 

57def revoke_token_by_ref(*, token_entry: TokenEntry, db: Session) -> None: 

58 token_entry._revoked_at = dt.datetime.now(tz=UTC) 

59 db.commit()