Setting the file. One moment. 67.2
Agentic Memory · references
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
def validate_s3_uri
— line 108
This file
- Number
- 67.18
- Position
- 18 of 19
- Type
- Python
- Size
- 6 KB
- Lines
- 191
scripts/input_validator.py
Python·191 lines·6 KB
import
re
18import sys
19
20
21class ValidationError(Exception):
22 def __init__(self, errors: list):
23 self.errors = errors
24 super().__init__(self.report())
25
26 def report(self) -> str:
27 return "\n".join(f" ❌ {e}" for e in self.errors)
28
29
30# Patterns
31CLUSTER_ID_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9\-]{0,62}$")
32GRAPH_ID_RE = re.compile(r"^g-[a-z0-9]{10,}$")
33REGION_RE = re.compile(r"^[a-z]{2}(-[a-z]+)+-\d+$")
34VPC_ID_RE = re.compile(r"^vpc-[0-9a-f]{8,17}$")
35SUBNET_ID_RE = re.compile(r"^subnet-[0-9a-f]{8,17}$")
36SG_ID_RE = re.compile(r"^sg-[0-9a-f]{8,17}$")
37KMS_KEY_RE = re.compile(
38 r"^(arn:aws(?:-cn|-us-gov)?:kms:[a-z0-9\-]+:\d{12}:key/[0-9a-f\-]+|[0-9a-f\-]{36})$"
39)
40IAM_ROLE_RE = re.compile(r"^arn:aws(?:-cn|-us-gov)?:iam::\d{12}:role/.+$")
41S3_URI_RE = re.compile(r"^s3://[a-z0-9][a-z0-9.\-]{1,61}[a-z0-9](/.+)?$")
42SNAPSHOT_ID_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9\-]{0,254}$")
43
44MIN_PROVISIONED_MEMORY = 16 # m-NCU (CreateGraph API minimum)
45MAX_PROVISIONED_MEMORY = 24576 # m-NCU (CreateGraph API maximum)
46
47
48def validate_cluster_id(value: str) -> str | None:
49 if not CLUSTER_ID_RE.match(value):
50 return f"cluster_id '{value}' invalid. Must start with letter, alphanumeric + hyphens, max 63 chars."
51 return None
52
53
54def validate_graph_id(value: str) -> str | None:
55 if not GRAPH_ID_RE.match(value):
56 return f"graph_id '{value}' invalid. Must match 'g-' followed by lowercase alphanumeric (e.g., g-abc123def4)."
57 return None
58
59
60def validate_region(value: str) -> str | None:
61 if not REGION_RE.match(value):
62 return f"region '{value}' invalid. Expected format: us-east-1, eu-west-2, etc."
63 return None
64
65
66def validate_vpc_id(value: str) -> str | None:
67 if not VPC_ID_RE.match(value):
68 return f"vpc_id '{value}' invalid. Expected format: vpc-0abc12345def67890."
69 return None
70
71
72def validate_subnet_ids(values) -> list:
73 if isinstance(values, str):
74 values = [v.strip() for v in values.split(",")]
75 errors = []
76 for v in values:
77 if not SUBNET_ID_RE.match(v):
78 errors.append(f"subnet_id '{v}' invalid. Expected format: subnet-0abc12345def67890.")
79 return errors
80
81
82def validate_security_group_ids(values) -> list:
83 if isinstance(values, str):
84 values = [v.strip() for v in values.split(",")]
85 errors = []
86 for v in values:
87 if not SG_ID_RE.match(v):
88 errors.append(
89 f"security_group_id '{v}' invalid. Expected format: sg-0abc12345def67890."
90 )
91 return errors
92
93
94def validate_kms_key(value: str) -> str | None:
95 if not KMS_KEY_RE.match(value):
96 return f"kms_key '{value}' invalid. Expected KMS key ARN or UUID."
97 return None
98
99
100def validate_iam_role(value: str) -> str | None:
101 if not IAM_ROLE_RE.match(value):
102 return (
103 f"iam_role '{value}' invalid. Expected format: arn:aws:iam::123456789012:role/RoleName."
104 )
105 return None
106
107
108def validate_s3_uri(value: str) -> str | None:
109 if not S3_URI_RE.match(value):
110 return f"s3_uri '{value}' invalid. Expected format: s3://bucket-name/path/."
111 return None
112
113
114def validate_snapshot_id(value: str) -> str | None:
115 if not SNAPSHOT_ID_RE.match(value):
116 return f"snapshot_id '{value}' invalid. Must start with letter, max 255 chars."
117 return None
118
119
120def validate_provisioned_memory(value) -> str | None:
121 try:
122 mem = int(value)
123 except (ValueError, TypeError):
124 return f"provisioned_memory '{value}' invalid. Must be an integer."
125 if mem < MIN_PROVISIONED_MEMORY or mem > MAX_PROVISIONED_MEMORY:
126 return f"provisioned_memory {mem} invalid. Valid range: {MIN_PROVISIONED_MEMORY}-{MAX_PROVISIONED_MEMORY} m-NCU."
127 return None
128
129
130VALIDATORS = {
131 "cluster_id": validate_cluster_id,
132 "graph_id": validate_graph_id,
133 "region": validate_region,
134 "vpc_id": validate_vpc_id,
135 "kms_key": validate_kms_key,
136 "iam_role": validate_iam_role,
137 "s3_uri": validate_s3_uri,
138 "snapshot_id": validate_snapshot_id,
139 "provisioned_memory": validate_provisioned_memory,
140}
141
142LIST_VALIDATORS = {
143 "subnet_ids": validate_subnet_ids,
144 "security_group_ids": validate_security_group_ids,
145}
146
147
148def validate_all(**kwargs) -> None:
149 errors = []
150 for key, value in kwargs.items():
151 if value is None:
152 continue
153 if key in VALIDATORS:
154 err = VALIDATORS[key](str(value))
155 if err:
156 errors.append(err)
157 elif key in LIST_VALIDATORS:
158 errors.extend(LIST_VALIDATORS[key](value))
159 else:
160 errors.append(
161 f"Unknown validator key: '{key}'. Valid keys: {sorted(list(VALIDATORS) + list(LIST_VALIDATORS))}"
162 )
163 if errors:
164 raise ValidationError(errors)
165
166
167def main():
168 if "--help" in sys.argv or len(sys.argv) < 2:
169 print("Usage: python3 scripts/input_validator.py key=value [key=value ...]")
170 print(f"Valid keys: {sorted(list(VALIDATORS) + list(LIST_VALIDATORS))}")
171 sys.exit(0)
172
173 kwargs = {}
174 for arg in sys.argv[1:]:
175 if "=" not in arg:
176 print(f"Invalid argument: {arg}. Use key=value format.")
177 sys.exit(1)
178 key, value = arg.split("=", 1)
179 kwargs[key] = value
180
181 try:
182 validate_all(**kwargs)
183 print("✅ All validations passed.")
184 except ValidationError as e:
185 print("Validation failed:")
186 print(e.report())
187 sys.exit(1)
188
189
190if __name__ == "__main__":
191 main()