Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/schemas/security.py: 85%

1from datetime import datetime, timedelta 

2from enum import Enum 

3from typing import Optional 

4from uuid import UUID, uuid4 

5 

6from pydantic import BaseModel, Field, validator 

7 

8PEER_TYPE_DESCRIPTION = ( 

9 "The peer type we generate access token for, i.e: opal client, data provider, etc." 

10) 

11TTL_DESCRIPTION = ( 

12 "Token lifetime (timedelta), can accept duration in seconds or ISO_8601 format." 

13 + " see: https://en.wikipedia.org/wiki/ISO_8601#Durations" 

14) 

15CLAIMS_DESCRIPTION = "extra claims to attach to the jwt" 

16 

17 

18class PeerType(str, Enum): 

19 client = "client" 

20 datasource = "datasource" 

21 listener = "listener" 

22 

23 

24class AccessTokenRequest(BaseModel): 

25 """A request to generate an access token to opal server.""" 

26 

27 id: UUID = Field(default_factory=uuid4) 

28 type: PeerType = Field(PeerType.client, description=PEER_TYPE_DESCRIPTION) 

29 ttl: timedelta = Field(timedelta(days=365), description=TTL_DESCRIPTION) 

30 claims: dict = Field({}, description=CLAIMS_DESCRIPTION) 

31 

32 @validator("type") 

33 def force_enum(cls, v): 

34 if isinstance(v, str): 34 ↛ 36line 34 didn't jump to line 36 because the condition on line 34 was always true

35 return PeerType(v) 

36 if isinstance(v, PeerType): 

37 return v 

38 raise ValueError(f"invalid value: {v}") 

39 

40 class Config: 

41 use_enum_values = True 

42 allow_population_by_field_name = True 

43 

44 

45class TokenDetails(BaseModel): 

46 id: UUID 

47 type: PeerType = Field(PeerType.client, description=PEER_TYPE_DESCRIPTION) 

48 expired: datetime 

49 claims: dict 

50 

51 

52class AccessToken(BaseModel): 

53 token: str 

54 type: str = "bearer" 

55 details: Optional[TokenDetails]