typing lints, environ

Author: ryan-williams

scripts/instance-runtime.py

@@ -14,9 +14,8 @@
 import subprocess
 import sys
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timezone
 from functools import partial
-from typing import Dict, List, Optional
 
 from dateutil import parser as date_parser
 
@@ -34,7 +33,7 @@
 
 err = partial(print, file=sys.stderr)
 
-def run_command(cmd: List[str]) -> Optional[str]:
+def run_command(cmd: list[str]) -> str | None:
     """Run a command and return output, or None on error."""
     try:
         result = subprocess.run(cmd, capture_output=True, text=True, check=True)
@@ -49,7 +48,7 @@ def run_command(cmd: List[str]) -> Optional[str]:
 
 
 
-def get_log_streams(instance_id: str, log_group: str = None) -> List[Dict]:
+def get_log_streams(instance_id: str, log_group: str = None) -> list[dict]:
     if log_group is None:
         log_group = DEFAULT_CLOUDWATCH_LOG_GROUP
     """Get CloudWatch log streams for an instance."""
@@ -69,7 +68,7 @@ def get_log_streams(instance_id: str, log_group: str = None) -> List[Dict]:
     return []
 
 
-def get_log_events(log_group: str, log_stream: str, limit: int = 100, start_from_head: bool = False) -> List[Dict]:
+def get_log_events(log_group: str, log_stream: str, limit: int = 100, start_from_head: bool = False) -> list[dict]:
     """Get events from a CloudWatch log stream."""
     cmd = [
         "aws", "logs", "get-log-events",
@@ -91,7 +90,7 @@ def get_log_events(log_group: str, log_stream: str, limit: int = 100, start_from
     return []
 
 
-def parse_timestamp(ts_str: str) -> Optional[datetime]:
+def parse_timestamp(ts_str: str) -> datetime | None:
     """Parse various timestamp formats and ensure timezone is set."""
     try:
         dt = date_parser.parse(ts_str)
@@ -103,7 +102,7 @@ def parse_timestamp(ts_str: str) -> Optional[datetime]:
         return None
 
 
-def extract_timestamp_from_log(message: str) -> Optional[datetime]:
+def extract_timestamp_from_log(message: str) -> datetime | None:
     """Extract timestamp from log message."""
     # Pattern: [Thu Aug 14 00:29:25 UTC 2025]
     match = re.search(r'\[([^]]+UTC \d{4})\]', message)
@@ -118,7 +117,7 @@ def extract_timestamp_from_log(message: str) -> Optional[datetime]:
     return None
 
 
-def analyze_instance(instance_id: str, log_group: str = None) -> Dict:
+def analyze_instance(instance_id: str, log_group: str = None) -> dict:
     if log_group is None:
         log_group = DEFAULT_CLOUDWATCH_LOG_GROUP
     """Analyze runtime and job execution for an instance."""
@@ -307,7 +306,7 @@ def analyze_instance(instance_id: str, log_group: str = None) -> Dict:
     return result
 
 
-def get_instances_from_github_url(url: str) -> List[str]:
+def get_instances_from_github_url(url: str) -> list[str]:
     """Extract instance IDs from a GitHub Actions URL."""
     # Parse the URL
     match = re.match(r'https://github\.com/([^/]+)/([^/]+)/actions/runs/(\d+)(?:/job/(\d+))?', url)

src/ec2_gha/__main__.py

@@ -17,17 +17,17 @@
 
 
 def main():
-    env = dict(os.environ)
+    env = dict(environ)
     required = ["GH_PAT", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]
     # Check that everything exists
     check_required(env, required)
     # Timeout for waiting for runner to register with GitHub
-    timeout_str = os.environ.get("INPUT_RUNNER_REGISTRATION_TIMEOUT", "").strip()
+    timeout_str = environ.get("INPUT_RUNNER_REGISTRATION_TIMEOUT", "").strip()
     timeout = int(timeout_str) if timeout_str else int(RUNNER_REGISTRATION_TIMEOUT)
 
-    token = os.environ["GH_PAT"]
+    token = environ["GH_PAT"]
     # Make a copy of environment variables for immutability
-    env = dict(os.environ)
+    env = dict(environ)
 
     builder = (
         EnvVarBuilder(env)

src/ec2_gha/start.py

@@ -135,16 +135,16 @@ def _build_aws_params(self, user_data_params: dict, idx: int = None) -> dict:
             template_vars = {}
 
             # Get repository name (just the basename)
-            if os.environ.get("GITHUB_REPOSITORY"):
-                template_vars["repo"] = os.environ["GITHUB_REPOSITORY"].split("/")[-1]
+            if environ.get("GITHUB_REPOSITORY"):
+                template_vars["repo"] = environ["GITHUB_REPOSITORY"].split("/")[-1]
             else:
                 template_vars["repo"] = "unknown"
 
             # Get workflow full name (e.g., "Test pip install")
-            template_vars["workflow"] = os.environ.get("GITHUB_WORKFLOW", "unknown")
+            template_vars["workflow"] = environ.get("GITHUB_WORKFLOW", "unknown")
 
             # Get workflow filename stem and ref from GITHUB_WORKFLOW_REF
-            workflow_ref = os.environ.get("GITHUB_WORKFLOW_REF", "")
+            workflow_ref = environ.get("GITHUB_WORKFLOW_REF", "")
             if workflow_ref:
                 import re
                 # Extract filename and ref from path like "owner/repo/.github/workflows/test.yml@ref"
@@ -168,7 +168,7 @@ def _build_aws_params(self, user_data_params: dict, idx: int = None) -> dict:
                 template_vars["ref"] = "unknown"
 
             # Get run number
-            template_vars["run_number"] = os.environ.get("GITHUB_RUN_NUMBER", "unknown")
+            template_vars["run_number"] = environ.get("GITHUB_RUN_NUMBER", "unknown")
 
             # Add instance index if provided (for multi-instance launches)
             if idx is not None:
@@ -182,16 +182,16 @@ def _build_aws_params(self, user_data_params: dict, idx: int = None) -> dict:
             default_tags.append({"Key": "Name", "Value": name_value})
 
         # Add repository tag if available
-        if "Repository" not in existing_keys and os.environ.get("GITHUB_REPOSITORY"):
-            default_tags.append({"Key": "Repository", "Value": os.environ["GITHUB_REPOSITORY"]})
+        if "Repository" not in existing_keys and environ.get("GITHUB_REPOSITORY"):
+            default_tags.append({"Key": "Repository", "Value": environ["GITHUB_REPOSITORY"]})
 
         # Add workflow tag if available
-        if "Workflow" not in existing_keys and os.environ.get("GITHUB_WORKFLOW"):
-            default_tags.append({"Key": "Workflow", "Value": os.environ["GITHUB_WORKFLOW"]})
+        if "Workflow" not in existing_keys and environ.get("GITHUB_WORKFLOW"):
+            default_tags.append({"Key": "Workflow", "Value": environ["GITHUB_WORKFLOW"]})
 
         # Add run URL tag if available
-        if "URL" not in existing_keys and os.environ.get("GITHUB_SERVER_URL") and os.environ.get("GITHUB_REPOSITORY") and os.environ.get("GITHUB_RUN_ID"):
-            gha_url = f"{os.environ['GITHUB_SERVER_URL']}/{os.environ['GITHUB_REPOSITORY']}/actions/runs/{os.environ['GITHUB_RUN_ID']}"
+        if "URL" not in existing_keys and environ.get("GITHUB_SERVER_URL") and environ.get("GITHUB_REPOSITORY") and environ.get("GITHUB_RUN_ID"):
+            gha_url = f"{environ['GITHUB_SERVER_URL']}/{environ['GITHUB_REPOSITORY']}/actions/runs/{environ['GITHUB_RUN_ID']}"
             default_tags.append({"Key": "URL", "Value": gha_url})
 
         # Combine user tags with default tags