Ruff

Author: ssjunnebo

dataflow_transfer/dataflow_transfer.py

@@ -2,7 +2,7 @@
 import time
 
 from dataflow_transfer.run_classes.registry import RUN_CLASS_REGISTRY
-from dataflow_transfer.utils.filesystem import get_run_dir, find_runs
+from dataflow_transfer.utils.filesystem import find_runs, get_run_dir
 
 logger = logging.getLogger(__name__)
 

dataflow_transfer/run_classes/generic_runs.py

@@ -1,9 +1,10 @@
-import os
 import logging
+import os
 import re
 from datetime import datetime
-from dataflow_transfer.utils.statusdb import StatusdbSession
+
 import dataflow_transfer.utils.filesystem as fs
+from dataflow_transfer.utils.statusdb import StatusdbSession
 
 logger = logging.getLogger(__name__)
 

dataflow_transfer/run_classes/illumina_runs.py

@@ -1,4 +1,5 @@
 from dataflow_transfer.run_classes.generic_runs import Run
+
 from .registry import register_run_class
 
 

dataflow_transfer/tests/test_run_classes.py

@@ -1,8 +1,9 @@
 import os
+
 import pytest
 
+from dataflow_transfer.run_classes import generic_runs, illumina_runs
 
-from dataflow_transfer.run_classes import illumina_runs, generic_runs
 # TODO: add tests for ONT and ELEMENT runs when those are implemented
 
 

dataflow_transfer/utils/filesystem.py

@@ -1,9 +1,10 @@
 import json
 import logging
 import os
-import xmltodict
 import subprocess
 
+import xmltodict
+
 logger = logging.getLogger(__name__)
 
 
@@ -40,9 +41,7 @@ def rsync_is_running(src):
 def submit_background_process(command_str: str):
     """Submit a command string as a background process."""
 
-    background_process = subprocess.Popen(
-        command_str, stdout=subprocess.PIPE, shell=True
-    )
+    subprocess.Popen(command_str, stdout=subprocess.PIPE, shell=True)
 
 
 def parse_metadata_files(files):
@@ -52,10 +51,10 @@ def parse_metadata_files(files):
     for file_path in files:
         try:
             if file_path.endswith(".json"):
-                with open(file_path, "r") as f:
+                with open(file_path) as f:
                     metadata[os.path.basename(file_path)] = json.load(f)
             elif file_path.endswith(".xml"):
-                with open(file_path, "r") as f:
+                with open(file_path) as f:
                     xml_content = xmltodict.parse(
                         f.read(), attr_prefix="", cdata_key="text"
                     )
@@ -71,10 +70,10 @@ def parse_metadata_files(files):
 
 
 def check_exit_status(file_path):
-    """Check the exit status from a given file. 
+    """Check the exit status from a given file.
     Return True if exit code is 0, else False."""
     if os.path.exists(file_path):
-        with open(file_path, "r") as f:
+        with open(file_path) as f:
             exit_code = f.read().strip()
             if exit_code == "0":
                 return True