Coverage for /usr/local/lib/python3.12/site-packages/prefect/blocks/system.py: 53%

26 statements  

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

1from __future__ import annotations 1a

2 

3import json 1a

4from typing import Annotated, Generic, TypeVar, Union 1a

5 

6from pydantic import ( 1a

7 Field, 

8 HttpUrl, 

9 JsonValue, 

10 SecretStr, 

11 StrictStr, 

12 field_validator, 

13) 

14from pydantic import Secret as PydanticSecret 1a

15 

16from prefect.blocks.core import Block 1a

17 

18_SecretValueType = Union[ 1a

19 Annotated[StrictStr, Field(title="string")], 

20 Annotated[JsonValue, Field(title="JSON")], 

21] 

22 

23T = TypeVar("T", bound=_SecretValueType) 1a

24 

25 

26class Secret(Block, Generic[T]): 1a

27 """ 

28 A block that represents a secret value. The value stored in this block will be obfuscated when 

29 this block is viewed or edited in the UI. 

30 

31 Attributes: 

32 value: A value that should be kept secret. 

33 

34 Example: 

35 ```python 

36 from prefect.blocks.system import Secret 

37 

38 Secret(value="sk-1234567890").save("BLOCK_NAME", overwrite=True) 

39 

40 secret_block = Secret.load("BLOCK_NAME") 

41 

42 # Access the stored secret 

43 secret_block.get() 

44 ``` 

45 """ 

46 

47 _logo_url = HttpUrl( 1a

48 "https://cdn.sanity.io/images/3ugk85nk/production/c6f20e556dd16effda9df16551feecfb5822092b-48x48.png" 

49 ) 

50 _documentation_url = HttpUrl("https://docs.prefect.io/latest/develop/blocks") 1a

51 _description = "A block that represents a secret value. The value stored in this block will be obfuscated when this block is viewed or edited in the UI." 1a

52 

53 value: Union[SecretStr, PydanticSecret[T]] = Field( 1a

54 default=..., 

55 description="A value that should be kept secret.", 

56 examples=["sk-1234567890", {"username": "johndoe", "password": "s3cr3t"}], 

57 json_schema_extra={ 

58 "writeOnly": True, 

59 "format": "password", 

60 }, 

61 ) 

62 

63 @field_validator("value", mode="before") 1a

64 def validate_value( 1a

65 cls, value: Union[T, SecretStr, PydanticSecret[T]] 

66 ) -> Union[SecretStr, PydanticSecret[T]]: 

67 if isinstance(value, (PydanticSecret, SecretStr)): 

68 return value 

69 else: 

70 return PydanticSecret[type(value)](value) 

71 

72 def get(self) -> T | str: 1a

73 value = self.value.get_secret_value() 

74 try: 

75 if isinstance(value, (str)): 

76 return json.loads(value) 

77 return value 

78 except (TypeError, json.JSONDecodeError): 

79 return value