create AWS infra

src/aws/cloudformation.py

@@ -0,0 +1,22 @@
+from typing import Any
+
+import boto3
+from botocore.exceptions import ClientError
+
+from src.infra.provisioning import STACK_NAME
+
+
+def stack_status(region: str, profile: str) -> dict[str, Any] | None:
+    client = boto3.Session(profile_name=profile, region_name=region).client("cloudformation")
+    try:
+        stack = client.describe_stacks(StackName=STACK_NAME)["Stacks"][0]
+    except ClientError as e:
+        message = e.response.get("Error", {}).get("Message", "")
+        if "does not exist" in message:
+            return None
+        raise
+    return {
+        "name": stack["StackName"],
+        "status": stack["StackStatus"],
+        "outputs": {item["OutputKey"]: item.get("OutputValue", "") for item in stack.get("Outputs", [])},
+    }

src/aws/identity.py

@@ -0,0 +1,5 @@
+import boto3
+
+
+def account_id(region: str, profile: str) -> str:
+    return boto3.Session(profile_name=profile, region_name=region).client("sts").get_caller_identity()["Account"]

src/aws/mlflow.py

@@ -0,0 +1,19 @@
+from typing import Any, cast
+
+import boto3
+from botocore.exceptions import ClientError
+
+
+def describe_tracking_server(region: str, profile: str, name: str) -> dict[str, Any] | None:
+    client = boto3.Session(profile_name=profile, region_name=region).client("sagemaker")
+    try:
+        return cast(dict[str, Any], client.describe_mlflow_tracking_server(TrackingServerName=name))
+    except ClientError as e:
+        code = e.response.get("Error", {}).get("Code", "")
+        message = e.response.get("Error", {}).get("Message", "")
+        if (
+            code in {"ResourceNotFound", "ResourceNotFoundException", "ValidationException"}
+            or "not found" in message.lower()
+        ):
+            return None
+        raise