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

1from datetime import datetime 

2from enum import Enum 

3from typing import Any, Dict, List, Optional 

4 

5from pydantic import BaseModel, Field, root_validator 

6 

7 

8class TransactionType(str, Enum): 

9 policy = "policy" 

10 data = "data" 

11 

12 

13class RemoteStatus(BaseModel): 

14 remote_url: str = Field(None, description="Url of remote data/policy source") 

15 succeed: bool = Field(True, description="Is request succeed") 

16 error: str = Field(None, description="If failed contains the type of exception") 

17 

18 

19class StoreTransaction(BaseModel): 

20 """Represents a transaction of policy or data to OPA.""" 

21 

22 id: str = Field(..., description="The id of the transaction") 

23 actions: List[str] = Field( 

24 ..., description="The write actions performed as part of the transaction" 

25 ) 

26 transaction_type: TransactionType = Field( 

27 None, description="Type of transaction,is it data/policy transaction" 

28 ) 

29 success: bool = Field( 

30 False, description="Whether or not the transaction was successful" 

31 ) 

32 error: str = Field( 

33 "", description="Error message in case of failure, defaults to empty string" 

34 ) 

35 creation_time: str = Field( 

36 None, description="Creation time for this store transaction" 

37 ) 

38 end_time: str = Field(None, description="Finish time for this store transaction") 

39 remotes_status: List[RemoteStatus] = Field( 

40 None, 

41 description="List of the remote sources for this transaction and their status", 

42 ) 

43 

44 

45class JSONPatchAction(BaseModel): 

46 """Abstract base class for JSON patch actions (RFC 6902)""" 

47 

48 op: str = Field(..., description="patch action to perform") 

49 path: str = Field(..., description="target location in modified json") 

50 value: Optional[Any] = Field( 

51 None, description="json document, the operand of the action" 

52 ) 

53 from_field: Optional[str] = Field( 

54 None, description="source location in json", alias="from" 

55 ) 

56 

57 @root_validator 

58 def value_must_be_present(cls, values): 

59 if values.get("op") in ["add", "replace"] and values.get("value") is None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 raise TypeError("'value' must be present when op is either add or replace") 

61 return values 

62 

63 

64class ArrayAppendAction(JSONPatchAction): 

65 op: str = Field("add", description="add action -> adds to the array") 

66 path: str = Field("-", description="dash marks the last index of an array")