Coverage for /usr/local/lib/python3.12/site-packages/prefect/types/names.py: 51%

59 statements  

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

1from __future__ import annotations 1a

2 

3import re 1a

4from functools import partial 1a

5from typing import Annotated, overload 1a

6 

7from pydantic import AfterValidator, BeforeValidator, Field 1a

8 

9LOWERCASE_LETTERS_NUMBERS_AND_DASHES_ONLY_REGEX = "^[a-z0-9-]*$" 1a

10LOWERCASE_LETTERS_NUMBERS_AND_UNDERSCORES_REGEX = "^[a-z0-9_]*$" 1a

11LOWERCASE_LETTERS_NUMBERS_AND_DASHES_OR_UNDERSCORES_REGEX = "^[a-z0-9-_]*$" 1a

12 

13 

14@overload 1a

15def raise_on_name_alphanumeric_dashes_only( 15 ↛ exitline 15 didn't return from function 'raise_on_name_alphanumeric_dashes_only' because 1a

16 value: str, field_name: str = ... 

17) -> str: ... 

18 

19 

20@overload 1a

21def raise_on_name_alphanumeric_dashes_only( 21 ↛ exitline 21 didn't return from function 'raise_on_name_alphanumeric_dashes_only' because 1a

22 value: None, field_name: str = ... 

23) -> None: ... 

24 

25 

26def raise_on_name_alphanumeric_dashes_only( 1a

27 value: str | None, field_name: str = "value" 

28) -> str | None: 

29 if value is not None and not bool( 

30 re.match(LOWERCASE_LETTERS_NUMBERS_AND_DASHES_ONLY_REGEX, value) 

31 ): 

32 raise ValueError( 

33 f"{field_name} must only contain lowercase letters, numbers, and dashes." 

34 ) 

35 return value 

36 

37 

38@overload 1a

39def raise_on_name_alphanumeric_underscores_only( 39 ↛ exitline 39 didn't return from function 'raise_on_name_alphanumeric_underscores_only' because 1a

40 value: str, field_name: str = ... 

41) -> str: ... 

42 

43 

44@overload 1a

45def raise_on_name_alphanumeric_underscores_only( 45 ↛ exitline 45 didn't return from function 'raise_on_name_alphanumeric_underscores_only' because 1a

46 value: None, field_name: str = ... 

47) -> None: ... 

48 

49 

50def raise_on_name_alphanumeric_underscores_only( 1a

51 value: str | None, field_name: str = "value" 

52) -> str | None: 

53 if value is not None and not re.match( 

54 LOWERCASE_LETTERS_NUMBERS_AND_UNDERSCORES_REGEX, value 

55 ): 

56 raise ValueError( 

57 f"{field_name} must only contain lowercase letters, numbers, and" 

58 " underscores." 

59 ) 

60 return value 

61 

62 

63def raise_on_name_alphanumeric_dashes_underscores_only( 1a

64 value: str, field_name: str = "value" 

65) -> str: 

66 if not re.match(LOWERCASE_LETTERS_NUMBERS_AND_DASHES_OR_UNDERSCORES_REGEX, value): 

67 raise ValueError( 

68 f"{field_name} must only contain lowercase letters, numbers, and" 

69 " dashes or underscores." 

70 ) 

71 return value 

72 

73 

74BANNED_CHARACTERS = ["/", "%", "&", ">", "<"] 1a

75 

76WITHOUT_BANNED_CHARACTERS = r"^[^" + "".join(BANNED_CHARACTERS) + "]+$" 1a

77Name = Annotated[str, Field(pattern=WITHOUT_BANNED_CHARACTERS)] 1a

78 

79WITHOUT_BANNED_CHARACTERS_EMPTY_OK = r"^[^" + "".join(BANNED_CHARACTERS) + "]*$" 1a

80NameOrEmpty = Annotated[str, Field(pattern=WITHOUT_BANNED_CHARACTERS_EMPTY_OK)] 1a

81 

82 

83def non_emptyish(value: str) -> str: 1a

84 if not value.strip("' \""): 

85 raise ValueError("name cannot be an empty string") 

86 

87 return value 

88 

89 

90NonEmptyishName = Annotated[ 1a

91 str, 

92 Field(pattern=WITHOUT_BANNED_CHARACTERS), 

93 BeforeValidator(non_emptyish), 

94] 

95 

96 

97### specific names 

98 

99BlockDocumentName = Annotated[ 1a

100 Name, 

101 AfterValidator( 

102 partial( 

103 raise_on_name_alphanumeric_dashes_only, field_name="Block document name" 

104 ) 

105 ), 

106] 

107 

108 

109BlockTypeSlug = Annotated[ 1a

110 str, 

111 AfterValidator( 

112 partial(raise_on_name_alphanumeric_dashes_only, field_name="Block type slug") 

113 ), 

114] 

115 

116ArtifactKey = Annotated[ 1a

117 str, 

118 AfterValidator( 

119 partial(raise_on_name_alphanumeric_dashes_only, field_name="Artifact key") 

120 ), 

121] 

122 

123MAX_VARIABLE_NAME_LENGTH = 255 1a

124 

125 

126VariableName = Annotated[ 1a

127 str, 

128 AfterValidator( 

129 partial( 

130 raise_on_name_alphanumeric_dashes_underscores_only, 

131 field_name="Variable name", 

132 ) 

133 ), 

134 Field( 

135 max_length=MAX_VARIABLE_NAME_LENGTH, 

136 description="The name of the variable", 

137 examples=["my_variable"], 

138 ), 

139] 

140 

141 

142# URI validation 

143URI_REGEX = re.compile(r"^[a-z0-9]+://") 1a

144 

145 

146def validate_uri(value: str) -> str: 1a

147 """Validate that a string is a valid URI with lowercase protocol.""" 

148 if not URI_REGEX.match(value): 

149 raise ValueError( 

150 "Key must be a valid URI, e.g. storage://bucket/folder/asset.csv" 

151 ) 

152 return value 

153 

154 

155URILike = Annotated[ 1a

156 str, 

157 AfterValidator(validate_uri), 

158 Field( 

159 description="A URI-like string with a lowercase protocol", 

160 examples=["s3://bucket/folder/data.csv", "postgres://dbtable"], 

161 ), 

162] 

163 

164 

165MAX_ASSET_KEY_LENGTH = 512 1a

166 

167RESTRICTED_ASSET_CHARACTERS = [ 1a

168 "\n", 

169 "\r", 

170 "\t", 

171 "\0", 

172 " ", 

173 "#", 

174 "?", 

175 "&", 

176 "%", 

177 '"', 

178 "'", 

179 "<", 

180 ">", 

181 "[", 

182 "]", 

183 "{", 

184 "}", 

185 "|", 

186 "\\", 

187 "^", 

188 "`", 

189] 

190 

191 

192def validate_valid_asset_key(value: str) -> str: 1a

193 """Validate asset key with character restrictions and length limit.""" 

194 for char in RESTRICTED_ASSET_CHARACTERS: 

195 if char in value: 

196 raise ValueError(f"Asset key cannot contain '{char}'") 

197 

198 if len(value) > MAX_ASSET_KEY_LENGTH: 

199 raise ValueError(f"Asset key cannot exceed {MAX_ASSET_KEY_LENGTH} characters") 

200 

201 return validate_uri(value) 

202 

203 

204ValidAssetKey = Annotated[ 1a

205 str, 

206 AfterValidator(validate_valid_asset_key), 

207 Field( 

208 max_length=MAX_ASSET_KEY_LENGTH, 

209 description=f"A URI-like string with a lowercase protocol, restricted characters, and max {MAX_ASSET_KEY_LENGTH} characters", 

210 examples=["s3://bucket/folder/data.csv", "postgres://dbtable"], 

211 ), 

212]