make pylint happy

.pylintrc

@@ -106,7 +106,12 @@ disable=
     fixme,
     cyclic-import,
     import-error,
-    missing-module-docstring
+    missing-module-docstring,
+    consider-alternative-union-syntax,
+    unspecified-encoding,
+    consider-using-with,
+    inconsistent-return-statements
+
 
 
 [REPORTS]
@@ -518,7 +523,7 @@ preferred-modules=
 
 # Exceptions that will emit a warning when being caught. Defaults to
 # "Exception"
-overgeneral-exceptions=Exception
+overgeneral-exceptions=builtins.Exception
 
 
 [TYPING]

gpu_jobs.py

@@ -1,6 +1,10 @@
 import os
 from dotenv import load_dotenv
-from utils.subprocess_operations import get_piped_stdout, stdout_to_gpu_records, job_records_to_slack_message
+from utils.subprocess_operations import (
+    get_piped_stdout,
+    stdout_to_gpu_records,
+    job_records_to_slack_message,
+)
 from hooks.slack import send_slack_message
 
 load_dotenv()
@@ -18,7 +22,7 @@ def monitor():
     gpu_jobs = get_gpu_jobs()
     gpu_records = stdout_to_gpu_records(gpu_jobs)
     header = f"🔉 Currently, there are {len(gpu_records)} running GPU Jobs on Bunya.\n Here are some from our lab.\n\n"
-    filtered_gpu_records = list(filter(lambda x: x['USER'] in GPU_USERS, gpu_records))
+    filtered_gpu_records = list(filter(lambda x: x["USER"] in GPU_USERS, gpu_records))
     slack_message = job_records_to_slack_message(header, filtered_gpu_records)
     send_slack_message(slack_message, SLACK_WEBHOOK)
 

hooks/slack.py

@@ -1,6 +1,7 @@
-import requests
 import json
 
+import requests
+
 
 def send_slack_message(message: str, webhook: str) -> str:
     payload = {"text": message}

my_jobs.py

@@ -49,10 +49,8 @@ def monitor_my_jobs():
 
     if last_updated_job_records:
         current_jobs_records = list_my_job_records()
-        current_job_ids = set([record["JOBID"] for record in current_jobs_records])
-        last_updated_job_ids = set(
-            [record["JOBID"] for record in last_updated_job_records]
-        )
+        current_job_ids = {record["JOBID"] for record in current_jobs_records}
+        last_updated_job_ids = {record["JOBID"] for record in last_updated_job_records}
 
         if current_job_ids != last_updated_job_ids:
             new_job_ids = current_job_ids.difference(last_updated_job_ids)
@@ -64,15 +62,19 @@ def monitor_my_jobs():
                     if record["JOBID"] in new_job_ids
                 ]
 
-                slack_message = job_records_to_slack_message("🔉 Update: New Jobs\n", new_job_records)
+                slack_message = job_records_to_slack_message(
+                    "🔉 Update: New Jobs\n", new_job_records
+                )
                 send_slack_message(message=slack_message, webhook=SLACK_WEBHOOK)
             if finished_job_ids != set():
                 finished_job_records = [
                     record
                     for record in last_updated_job_records
                     if record["JOBID"] in finished_job_ids
                 ]
-                slack_message = job_records_to_slack_message("🔉 Update: Complete Jobs\n", finished_job_records)
+                slack_message = job_records_to_slack_message(
+                    "🔉 Update: Complete Jobs\n", finished_job_records
+                )
                 send_slack_message(message=slack_message, webhook=SLACK_WEBHOOK)
             write_job_records_to_json(current_jobs_records, JOB_FILE_PATH)
     else:

utils/subprocess_operations.py

@@ -14,8 +14,7 @@ def get_cmd_stdout(command: str) -> str:
 
     if result.returncode == 0:
         return result.stdout
-    else:
-        print(f"Command: {command} failed with error: {result.stderr}")
+    print(f"Command: {command} failed with error: {result.stderr}")
 
 
 def get_piped_stdout(main_command: str, piped_command: str) -> Optional[str]:
@@ -35,10 +34,10 @@ def get_piped_stdout(main_command: str, piped_command: str) -> Optional[str]:
 
         if piped_result.returncode == 0:
             return piped_result.stdout
-    else:
-        print(
-            f"Command: {main_command} | {piped_command} failed with error: {initial_command_result.stderr}"
-        )
+
+    print(
+        f"Command: {main_command} | {piped_command} failed with error: {initial_command_result.stderr}"
+    )
 
 
 def strip_spaces(l: list[str]) -> list[str]: