fix: typing strictness

auth0_to_hubspot/program.py

@@ -1,14 +1,14 @@
 """This program adds new Auth0 users to HubSpot as contacts."""
 
-from datetime import datetime, UTC
+from datetime import datetime, timedelta, UTC
 import os
 
 from autokitteh.auth0 import auth0_client
 from autokitteh.hubspot import hubspot_client
 from hubspot.crm.contacts import SimplePublicObjectInput
 
 
-LOOKUP_HOURS = int(os.getenv("HOURS"))
+LOOKUP_HOURS = int(os.getenv("HOURS") or "24")
 
 auth0 = auth0_client("auth0_conn")
 hubspot = hubspot_client("hubspot_conn")
@@ -29,7 +29,7 @@ def check_for_new_users(event):
 def _get_time_range(hours):
     """Calculate start and end times for user lookup."""
     now = datetime.now(UTC)
-    start_time = now - datetime.timedelta(hours=hours)
+    start_time = now - timedelta(hours=hours)
     return (start_time.isoformat() + "Z", now.isoformat() + "Z")
 
 

aws_health_to_slack/program.py

@@ -8,15 +8,14 @@
 from datetime import datetime, timedelta, UTC
 import json
 import os
-import re
 
 import autokitteh
 from autokitteh.aws import boto3_client
 from autokitteh.google import google_id, google_sheets_client
 from autokitteh.slack import slack_client
 
 
-url = os.getenv("GOOGLE_SHEET_URL")
+URL = os.getenv("GOOGLE_SHEET_URL") or ""
 
 
 def on_schedule(_):
@@ -44,7 +43,7 @@ def on_schedule(_):
 def _read_google_sheet() -> dict[str, str]:
     """Read mapping of project tags to Slack channels from Google Sheet."""
     sheets = google_sheets_client("google_sheets_connection").spreadsheets().values()
-    rows = sheets.get(spreadsheetId=google_id(url), range="A:B").execute()
+    rows = sheets.get(spreadsheetId=google_id(URL), range="A:B").execute()
     return {row[0].strip(): row[1].strip() for row in rows.get("values", [])}
 
 
@@ -59,7 +58,7 @@ def _aws_health_events() -> list[dict]:
     https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/health/client/describe_events_for_organization.html
     """
     try:
-        mins = int(re.match(r"(\d+)m", os.getenv("TRIGGER_INTERVAL")).group(1))
+        mins = int((os.getenv("TRIGGER_INTERVAL") or "1m")[:-1])
         prev_check = datetime.now(UTC) - timedelta(minutes=mins)
         filter = {"lastUpdatedTimes": [{"from": prev_check}]}
 
@@ -111,7 +110,7 @@ def _affected_aws_entities(events: list[dict]) -> list[dict]:
 
 def _post_slack_message(channel, project, entity: dict, affecting_events: list[dict]):
     if not channel:
-        print(f"Error: project {project!r} not found in {url}")
+        print(f"Error: project {project!r} not found in {URL}")
 
     text = f"This AWS resource:\n```\n{json.dumps(entity, indent=4)}\n```"
     text += "\nis affected by these AWS Health events:"

data_pipeline/pipeline.py

@@ -65,13 +65,17 @@ def insert_records(db_dsn, records):
 def parse_gpx(track_id, data):
     io = BytesIO(data)
     root = Xml.parse(io).getroot()
-    return [
-        {
-            "track_id": track_id,
-            "n": i,
-            "lat": float(elem.get("lat")),
-            "lng": float(elem.get("lon")),
-            "height": float(elem.findtext(".//")),
-        }
-        for i, elem in enumerate(root.findall(".//" + trkpt_tag))
-    ]
+    records = []
+
+    for i, elem in enumerate(root.findall(".//" + trkpt_tag)):
+        records.append(
+            {
+                "track_id": track_id,
+                "n": i,
+                "lat": float(elem.get("lat", "0")),
+                "lng": float(elem.get("lon", "0")),
+                "height": float(elem.findtext(".//") or "0"),
+            }
+        )
+
+    return records

hackernews/program.py

@@ -9,7 +9,7 @@
 
 
 API_URL = "http://hn.algolia.com/api/v1/search_by_date?tags=story&page=0&query="
-POLLING_INTERVAL_SECS = int(os.getenv("POLLING_INTERVAL_SECS"))
+POLLING_INTERVAL_SECS = int(os.getenv("POLLING_INTERVAL_SECS") or "120")
 
 slack = slack_client("slack_connection")
 

purrr/github_helper.py

@@ -5,6 +5,6 @@
 from autokitteh.github import github_client
 
 
-ORG_NAME = os.getenv("github_conn__target_name")
+ORG_NAME = os.getenv("github_conn__target_name") or ""
 
 shared_client = github_client("github_conn")

purrr/slack_channel.py

@@ -128,9 +128,9 @@ def add_users(channel_id: str, github_users: list[str]) -> None:
     if len(slack_users) > 1000:
         slack_users = slack_users[:1000]
 
-    users = ",".join(slack_users)
+    user_ids = ",".join(slack_users)
     try:
-        slack.conversations_invite(channel=channel_id, users=users, force=True)
+        slack.conversations_invite(channel=channel_id, users=user_ids, force=True)
     except SlackApiError as e:
         if e.response["error"] == "already_in_channel":
             return

purrr/users.py

@@ -1,7 +1,7 @@
 """User-related helper functions across GitHub and Slack."""
 
 from autokitteh.slack import slack_client
-import github
+from github import NamedUser, GithubException
 from slack_sdk.errors import SlackApiError
 
 import data_helper
@@ -106,7 +106,11 @@ def format_slack_user_for_github(slack_user_id: str) -> str:
         debug.log(f"Slack user <@{slack_user_id}>: `real_name` not found in profile")
         return "Someone"
 
-    users = [user for user in _github_users() if user.name.lower() == slack_name]
+    users = []
+    for user in _github_users():
+        if user.name and user.name.lower() == slack_name:
+            users.append(user)
+
     if len(users) == 1:
         github_ref = "@" + users[0].login
         data_helper.cache_github_reference(slack_user_id, github_ref)
@@ -121,11 +125,11 @@ def format_slack_user_for_github(slack_user_id: str) -> str:
     return profile["real_name"]
 
 
-def _github_users() -> list[github.NamedUser.NamedUser]:
+def _github_users() -> list[NamedUser.NamedUser]:
     """Return a list of all GitHub users in the organization."""
     try:
         return list(gh.get_organization(github_helper.ORG_NAME).get_members())
-    except github.GithubException as e:
+    except GithubException as e:
         error = "Failed to list GitHub members in the organization"
         debug.log(f"{error} `{github_helper.ORG_NAME}`:\n```{e}```")
         return []
@@ -217,8 +221,9 @@ def github_username_to_slack_user_id(github_username: str) -> str:
             profile.get("real_name_normalized", "").lower(),
         )
         if github_name in slack_names:
-            data_helper.cache_slack_user_id(github_username, user.id)
-            return user.id
+            slack_user_id = user.get("id", "")
+            data_helper.cache_slack_user_id(github_username, slack_user_id)
+            return slack_user_id
 
     # Optimization: cache unsuccessful results too (i.e. external users).
     debug.log(f"GitHub user {gh_user_link}: email & name not found in Slack")

room_reservation/reserve_room.py

@@ -32,7 +32,7 @@ def on_slack_slash_command(event):
         slack.chat_postMessage(channel=channel_id, text=err)
         return
 
-    user = slack.users_profile_get(user=data.user_id).get("profile")
+    user = slack.users_profile_get(user=data.user_id).get("profile", {})
 
     now = datetime.now(UTC)
     in_5_minutes = now + timedelta(minutes=5)