Coverage for docs / client-examples / python_boto3_example.py: 100.00%

1#!/usr/bin/env python3 

2""" 

3Example: Submit Kubernetes manifests to GCO API Gateway using Python boto3 

4 

5This example demonstrates how to authenticate with AWS IAM and submit manifests 

6to the GCO API Gateway using AWS SigV4 request signing. 

7 

8Requirements: 

9 pip install boto3 requests aws-requests-auth 

10 

11Usage: 

12 python python_boto3_example.py 

13""" 

14 

15import json 

16from pathlib import Path 

17from typing import Any 

18 

19import boto3 

20import requests 

21from aws_requests_auth.aws_auth import AWSRequestsAuth 

22 

23 

24def get_api_endpoint(region: str, project_name: str = "gco") -> str: 

25 """ 

26 Get the API Gateway endpoint URL from CloudFormation stack outputs. 

27 

28 Args: 

29 region: AWS region where API Gateway stack is deployed. 

30 project_name: Deployment project prefix (defaults to ``gco``). 

31 

32 Returns: 

33 API Gateway invoke URL 

34 """ 

35 cfn = boto3.client("cloudformation", region_name=region) 

36 

37 stack_name = f"{project_name}-api-gateway" 

38 response = cfn.describe_stacks(StackName=stack_name) 

39 outputs = response["Stacks"][0]["Outputs"] 

40 

41 for output in outputs: 

42 if output["OutputKey"] == "ApiEndpoint": 

43 # Remove trailing slash if present 

44 endpoint: str = output["OutputValue"] 

45 return endpoint.rstrip("/") 

46 

47 raise ValueError(f"ApiEndpoint not found in stack {stack_name}") 

48 

49 

50def create_aws_auth(api_host: str, region: str) -> AWSRequestsAuth: 

51 """ 

52 Create AWS SigV4 authentication for API Gateway requests. 

53 

54 Args: 

55 api_host: API Gateway host (e.g., 'abc123.execute-api.us-east-2.amazonaws.com') 

56 region: AWS region 

57 

58 Returns: 

59 AWSRequestsAuth object for request signing 

60 """ 

61 session = boto3.Session() 

62 credentials = session.get_credentials() 

63 if credentials is None: 

64 raise RuntimeError( 

65 "No AWS credentials are available. Configure a profile, environment credentials, " 

66 "or an IAM role before running this example." 

67 ) 

68 frozen = credentials.get_frozen_credentials() 

69 

70 return AWSRequestsAuth( 

71 aws_access_key=frozen.access_key, 

72 aws_secret_access_key=frozen.secret_key, 

73 aws_token=frozen.token, # For temporary credentials (STS, IAM roles) 

74 aws_host=api_host, 

75 aws_region=region, 

76 aws_service="execute-api", 

77 ) 

78 

79 

80def submit_manifests( 

81 api_endpoint: str, 

82 auth: AWSRequestsAuth, 

83 manifests: list[dict[str, Any]], 

84 namespace: str | None = None, 

85 dry_run: bool = False, 

86) -> dict[str, Any]: 

87 """ 

88 Submit Kubernetes manifests to the API Gateway. 

89 

90 Args: 

91 api_endpoint: API Gateway base URL 

92 auth: AWS authentication object 

93 manifests: List of Kubernetes manifests as dictionaries 

94 namespace: Default namespace for manifests without one specified 

95 dry_run: If True, validate without applying 

96 

97 Returns: 

98 API response as dictionary 

99 """ 

100 url = f"{api_endpoint}/api/v1/manifests" 

101 

102 payload = {"manifests": manifests, "dry_run": dry_run} 

103 

104 if namespace: 

105 payload["namespace"] = namespace 

106 

107 response = requests.post( 

108 url, json=payload, auth=auth, headers={"Content-Type": "application/json"}, timeout=30 

109 ) 

110 

111 response.raise_for_status() 

112 body: dict[str, Any] = response.json() 

113 return body 

114 

115 

116def get_health(api_endpoint: str, auth: AWSRequestsAuth) -> dict[str, Any]: 

117 """ 

118 Get cluster health status. 

119 

120 Args: 

121 api_endpoint: API Gateway base URL 

122 auth: AWS authentication object 

123 

124 Returns: 

125 API response as dictionary 

126 """ 

127 url = f"{api_endpoint}/api/v1/health" 

128 

129 response = requests.get(url, auth=auth, timeout=30) 

130 response.raise_for_status() 

131 body: dict[str, Any] = response.json() 

132 return body 

133 

134 

135def get_deployment_config() -> tuple[str, str]: 

136 """Return ``(project_name, API Gateway region)`` from ``cdk.json``. 

137 

138 The stock values are used when the file is unavailable or malformed. AWS 

139 credentials still come from boto3's normal provider chain, including 

140 ``AWS_PROFILE``, SSO, web identity, containers, and instance roles. 

141 """ 

142 cdk_json_path = Path(__file__).resolve().parents[2] / "cdk.json" 

143 try: 

144 with cdk_json_path.open(encoding="utf-8") as file: 

145 context = json.load(file).get("context", {}) 

146 project_name = str(context.get("project_name") or "gco") 

147 deployment_regions = context.get("deployment_regions", {}) 

148 api_region = str(deployment_regions.get("api_gateway") or "us-east-2") 

149 return project_name, api_region 

150 except OSError, AttributeError, ValueError: 

151 # OSError: no readable cdk.json; ValueError: not JSON; AttributeError: 

152 # valid JSON of the wrong shape (a list or scalar where a mapping was 

153 # expected has no ``.get``). 

154 return "gco", "us-east-2" 

155 

156 

157def main() -> None: 

158 project_name, api_region = get_deployment_config() 

159 stack_name = f"{project_name}-api-gateway" 

160 print(f"Using API Gateway region: {api_region}") 

161 

162 # Get API Gateway endpoint from CloudFormation 

163 print(f"Getting API Gateway endpoint from stack {stack_name}...") 

164 api_endpoint = get_api_endpoint(api_region, project_name) 

165 print(f"API Endpoint: {api_endpoint}") 

166 

167 # Extract host from endpoint URL 

168 api_host = api_endpoint.replace("https://", "").replace("http://", "").split("/")[0] 

169 

170 # Create AWS authentication 

171 print("Creating AWS SigV4 authentication...") 

172 auth = create_aws_auth(api_host, api_region) 

173 

174 # Example 1: Simple Kubernetes Job 

175 print("\n=== Example 1: Submit a simple Job ===") 

176 simple_job = { 

177 "apiVersion": "batch/v1", 

178 "kind": "Job", 

179 "metadata": {"name": "python-example-job", "namespace": "gco-jobs"}, 

180 "spec": { 

181 "template": { 

182 "spec": { 

183 "containers": [ 

184 { 

185 "name": "example", 

186 "image": "busybox:1.38.0", 

187 "command": ["echo", "Hello from GCO Python client!"], 

188 } 

189 ], 

190 "restartPolicy": "Never", 

191 } 

192 }, 

193 "backoffLimit": 3, 

194 }, 

195 } 

196 

197 try: 

198 result = submit_manifests(api_endpoint, auth, [simple_job]) 

199 print(f"Success: {json.dumps(result, indent=2)}") 

200 except requests.exceptions.HTTPError as e: 

201 print(f"Error: {e}") 

202 print(f"Response: {e.response.text}") 

203 

204 # Example 2: GPU Job with node selector for on-demand capacity 

205 print("\n=== Example 2: Submit a GPU Job (on-demand) ===") 

206 gpu_job = { 

207 "apiVersion": "batch/v1", 

208 "kind": "Job", 

209 "metadata": {"name": "gpu-python-job", "namespace": "gco-jobs"}, 

210 "spec": { 

211 "template": { 

212 "spec": { 

213 "containers": [ 

214 { 

215 "name": "gpu-example", 

216 "image": "nvidia/cuda:12.0-base", 

217 "command": ["nvidia-smi"], 

218 "resources": {"limits": {"nvidia.com/gpu": "1"}}, 

219 } 

220 ], 

221 "restartPolicy": "Never", 

222 "nodeSelector": {"karpenter.sh/capacity-type": "on-demand"}, 

223 "tolerations": [ 

224 {"key": "nvidia.com/gpu", "operator": "Exists", "effect": "NoSchedule"} 

225 ], 

226 } 

227 }, 

228 "backoffLimit": 3, 

229 }, 

230 } 

231 

232 print(f"GPU Job manifest: {json.dumps(gpu_job, indent=2)}") 

233 print("(Not submitting - uncomment to test with GPU nodes)") 

234 # Uncomment to submit: 

235 # try: 

236 # result = submit_manifests(api_endpoint, auth, [gpu_job]) 

237 # print(f"Success: {json.dumps(result, indent=2)}") 

238 # except requests.exceptions.HTTPError as e: 

239 # print(f"Error: {e}") 

240 # print(f"Response: {e.response.text}") 

241 

242 # Example 3: Multiple manifests at once 

243 print("\n=== Example 3: Submit multiple manifests ===") 

244 config_map = { 

245 "apiVersion": "v1", 

246 "kind": "ConfigMap", 

247 "metadata": {"name": "python-example-config", "namespace": "gco-jobs"}, 

248 "data": {"config.yaml": "key: value\nother: setting"}, 

249 } 

250 

251 config_reader_job = { 

252 "apiVersion": "batch/v1", 

253 "kind": "Job", 

254 "metadata": {"name": "config-reader-python-job", "namespace": "gco-jobs"}, 

255 "spec": { 

256 "template": { 

257 "spec": { 

258 "containers": [ 

259 { 

260 "name": "reader", 

261 "image": "busybox:1.38.0", 

262 "command": ["cat", "/config/config.yaml"], 

263 "volumeMounts": [{"name": "config-volume", "mountPath": "/config"}], 

264 } 

265 ], 

266 "volumes": [ 

267 {"name": "config-volume", "configMap": {"name": "python-example-config"}} 

268 ], 

269 "restartPolicy": "Never", 

270 } 

271 }, 

272 "backoffLimit": 3, 

273 }, 

274 } 

275 

276 try: 

277 result = submit_manifests(api_endpoint, auth, [config_map, config_reader_job]) 

278 print(f"Success: {json.dumps(result, indent=2)}") 

279 except requests.exceptions.HTTPError as e: 

280 print(f"Error: {e}") 

281 print(f"Response: {e.response.text}") 

282 

283 # Example 4: Dry run validation 

284 print("\n=== Example 4: Dry run validation ===") 

285 try: 

286 result = submit_manifests(api_endpoint, auth, [simple_job], dry_run=True) 

287 print(f"Dry run result: {json.dumps(result, indent=2)}") 

288 except requests.exceptions.HTTPError as e: 

289 print(f"Error: {e}") 

290 print(f"Response: {e.response.text}") 

291 

292 print("\n=== Examples Complete ===") 

293 print("\nKey points:") 

294 print("1. The API expects 'manifests' as a list of JSON objects (not YAML strings)") 

295 print("2. Each manifest must include apiVersion, kind, and metadata with name") 

296 print("3. Use aws-requests-auth for SigV4 signing with requests library") 

297 print("4. Use nodeSelector 'karpenter.sh/capacity-type' to control spot vs on-demand") 

298 

299 

300if __name__ == "__main__": 

301 main()