Coverage for /usr/local/lib/python3.12/site-packages/prefect/types/names.py: 66%
59 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1from __future__ import annotations 1a
3import re 1a
4from functools import partial 1a
5from typing import Annotated, overload 1a
7from pydantic import AfterValidator, BeforeValidator, Field 1a
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
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: ...
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: ...
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( 1bcd
30 re.match(LOWERCASE_LETTERS_NUMBERS_AND_DASHES_ONLY_REGEX, value)
31 ):
32 raise ValueError( 1bcd
33 f"{field_name} must only contain lowercase letters, numbers, and dashes."
34 )
35 return value 1bcd
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: ...
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: ...
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
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): 1bcd
67 raise ValueError( 1bcd
68 f"{field_name} must only contain lowercase letters, numbers, and"
69 " dashes or underscores."
70 )
71 return value 1bcd
74BANNED_CHARACTERS = ["/", "%", "&", ">", "<"] 1a
76WITHOUT_BANNED_CHARACTERS = r"^[^" + "".join(BANNED_CHARACTERS) + "]+$" 1a
77Name = Annotated[str, Field(pattern=WITHOUT_BANNED_CHARACTERS)] 1a
79WITHOUT_BANNED_CHARACTERS_EMPTY_OK = r"^[^" + "".join(BANNED_CHARACTERS) + "]*$" 1a
80NameOrEmpty = Annotated[str, Field(pattern=WITHOUT_BANNED_CHARACTERS_EMPTY_OK)] 1a
83def non_emptyish(value: str) -> str: 1a
84 if not value.strip("' \""): 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true1b
85 raise ValueError("name cannot be an empty string")
87 return value 1b
90NonEmptyishName = Annotated[ 1a
91 str,
92 Field(pattern=WITHOUT_BANNED_CHARACTERS),
93 BeforeValidator(non_emptyish),
94]
97### specific names
99BlockDocumentName = Annotated[ 1a
100 Name,
101 AfterValidator(
102 partial(
103 raise_on_name_alphanumeric_dashes_only, field_name="Block document name"
104 )
105 ),
106]
109BlockTypeSlug = Annotated[ 1a
110 str,
111 AfterValidator(
112 partial(raise_on_name_alphanumeric_dashes_only, field_name="Block type slug")
113 ),
114]
116ArtifactKey = Annotated[ 1a
117 str,
118 AfterValidator(
119 partial(raise_on_name_alphanumeric_dashes_only, field_name="Artifact key")
120 ),
121]
123MAX_VARIABLE_NAME_LENGTH = 255 1a
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]
142# URI validation
143URI_REGEX = re.compile(r"^[a-z0-9]+://") 1a
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
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]
165MAX_ASSET_KEY_LENGTH = 512 1a
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]
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}'")
198 if len(value) > MAX_ASSET_KEY_LENGTH:
199 raise ValueError(f"Asset key cannot exceed {MAX_ASSET_KEY_LENGTH} characters")
201 return validate_uri(value)
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]