replace most uses of os.path with pathlib

skare3_tools/config.py

@@ -72,6 +72,7 @@
 
 import json
 import os
+from pathlib import Path
 
 # this is just a default config. This gets saved in a file which can be modified later on.
 # If the file exists, this will be ignored unless explicitly resetting.
@@ -91,7 +92,7 @@
         ],
     },
     "organizations": ["sot", "acisops"],
-    "data_dir": "",
+    "data_dir": None,
 }
 
 
@@ -103,10 +104,10 @@
 
 
 def _app_data_dir_():
-    ska_data_dir = os.path.join(os.environ["SKA"], "data", "skare3", "skare3_data")
+    ska_data_dir = Path(os.environ["SKA"]) / "data" / "skare3" / "skare3_data"
     if "SKARE3_TOOLS_DATA" in os.environ:
         app_data_dir = os.environ["SKARE3_TOOLS_DATA"]
-    elif os.path.exists(ska_data_dir):
+    elif ska_data_dir.exists():
         app_data_dir = ska_data_dir
     else:
         app_data_dir = None
@@ -129,8 +130,8 @@ def init(config=None, reset=False):
         raise Exception(
             "Could not figure out where to place skare3_tools configuration"
         )
-    config_file = os.path.join(app_data_dir, "config.json")
-    exists = os.path.exists(config_file)
+    config_file = app_data_dir / "config.json"
+    exists = config_file.exists()
     if exists and not reset:
         with open(config_file) as f:
             CONFIG = json.load(f)
@@ -141,9 +142,8 @@ def init(config=None, reset=False):
         if reset:
             CONFIG = _DEFAULT_CONFIG.copy()
         if "data_dir" not in CONFIG or not CONFIG["data_dir"]:
-            CONFIG["data_dir"] = os.path.join(app_data_dir, "data")
-        if not os.path.exists(CONFIG["data_dir"]):
-            os.makedirs(CONFIG["data_dir"])
+            CONFIG["data_dir"] = app_data_dir.resolve() / "data"
+        CONFIG["data_dir"].mkdir(parents=True, exist_ok=True)
         with open(config_file, "w") as f:
             json.dump(CONFIG, f, indent=2)
 

skare3_tools/dashboard/views/test_log.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 
-import os
 import re
 
 from skare3_tools import config, dashboard
@@ -90,13 +89,13 @@ def test_log(path):
         return dashboard.get_template("error.html").render(
             title="404 Error", message="Run {run_id} not found".format(run_id=run_id)
         )
-    filename = os.path.join(
-        config.CONFIG["data_dir"],
-        "test_logs",
-        test_runs[0]["run_info"]["destination"],
-        filename,
+    filename = (
+        config.CONFIG["data_dir"]
+        / "test_logs"
+        / test_runs[0]["run_info"]["destination"]
+        / filename
     )
-    if not os.path.exists(filename) or not os.path.isfile(filename):
+    if not filename.exists() or not filename.is_file():
         return dashboard.get_template("error.html").render(
             title="404 Error", message="File {path} not found".format(path=path)
         )

skare3_tools/github/scripts/add_secrets.py

@@ -32,6 +32,7 @@
 import os
 import sys
 import time
+from pathlib import Path
 from urllib.parse import urlparse
 
 import yaml
@@ -163,7 +164,10 @@ def parser():
     parse = argparse.ArgumentParser(description=__doc__)
     parse.add_argument("repositories", nargs="+")
     parse.add_argument(
-        "--secrets", default="secrets.json", help="JSON file with all secrets"
+        "--secrets",
+        default="secrets.json",
+        help="JSON file with all secrets",
+        type=Path,
     )
     parse.add_argument("--user", required=True, help="Github user name")
     parse.add_argument(
@@ -190,7 +194,7 @@ def main():
         )
         the_parser.exit(2)
 
-    if not os.path.exists(args.secrets):
+    if not args.secrets.exists():
         logging.error(
             f"The secrets file {args.secrets} does not exist. "
             f"Run `{sys.argv[0]} -h` for help."

skare3_tools/packages.py

@@ -120,7 +120,7 @@ def json_cache(name, directory="", ignore=None, expires=None, update_policy=None
     import inspect
     from functools import wraps
 
-    directory = os.path.normpath(os.path.join(CONFIG["data_dir"], directory))
+    directory = CONFIG["data_dir"] / directory
     if not ignore:
         ignore = []
     if expires:
@@ -132,7 +132,7 @@ def decorator_cache(func, ignore_args=ignore, expiration=expires, name=name):
 
         def dir_access_ok(path):
             path = Path(path).resolve()
-            if os.path.exists(path):
+            if path.exists():
                 return os.access(path, os.W_OK)
             if path.parent != path:
                 return dir_access_ok(path.parent)
@@ -151,12 +151,12 @@ def wrapper(*args, update=False, **kwargs):
             filename = "{name}{arg_str}.json".format(name=name, arg_str=arg_str)
             # in an ideal world, filename would be completely sanitized... this world is not ideal.
             filename = filename.replace(os.sep, "-")
-            filename = os.path.join(directory, filename)
-            if expiration is not None and os.path.exists(filename):
-                m_time = datetime.datetime.fromtimestamp(os.path.getmtime(filename))
+            filename = directory / filename
+            if expiration is not None and filename.exists():
+                m_time = datetime.datetime.fromtimestamp(filename.stat().st_mtime)
                 update = update or (datetime.datetime.now() - m_time > expiration)
             result = None
-            if os.path.exists(filename):
+            if filename.exists():
                 with open(filename) as file:
                     result = json.load(file)
             if update_policy is not None and result is not None:
@@ -172,16 +172,13 @@ def wrapper(*args, update=False, **kwargs):
                 update = False
             if result is None or update:
                 result = func(*args, **kwargs)
-                directory_out = os.path.dirname(filename)
-                if not os.path.exists(directory_out):
-                    os.makedirs(directory_out)
+                filename.parent.mkdir(parents=True, exists_ok=True)
                 with open(filename, "w") as file:
                     json.dump(result, file)
             return result
 
         def clear_cache():
-            files = os.path.join(directory, "{name}*.json".format(name=name))
-            files = glob.glob(files)
+            files = list(directory.glob(f"{name}*.json"))
             if files:
                 subprocess.run(["rm"] + files, check=False)
 
@@ -198,10 +195,8 @@ def rm_cache_entry(*args, s=sig, **kwargs):
                     if a not in ignore_args
                 ]
             )
-            filename = os.path.join(
-                directory, "{name}{arg_str}.json".format(name=name, arg_str=arg_str)
-            )
-            if os.path.exists(filename):
+            filename = directory / f"{name}{arg_str}.json"
+            if filename.exists():
                 os.remove(filename)
 
         wrapper.rm_cache_entry = rm_cache_entry
@@ -211,11 +206,9 @@ def rm_cache_entry(*args, s=sig, **kwargs):
 
 
 def _ensure_skare3_local_repo(update=True):
-    repo_dir = os.path.join(CONFIG["data_dir"], "skare3")
-    parent = os.path.dirname(repo_dir)
-    if not os.path.exists(parent):
-        os.makedirs(parent)
-    if not os.path.exists(repo_dir):
+    repo_dir = CONFIG["data_dir"] / "skare3"
+    repo_dir.parent.mkdir(parents=True, exist_ok=True)
+    if not repo_dir.exists():
         _ = subprocess.run(
             ["git", "clone", "https://github.com/sot/skare3", repo_dir],
             cwd=CONFIG["data_dir"],
@@ -231,22 +224,20 @@ def _ensure_skare3_local_repo(update=True):
             stderr=subprocess.PIPE,
             check=False,
         )
-    assert os.path.exists(repo_dir)
+    assert repo_dir.exists()
 
 
 def _conda_package_list(update=True):
     _ensure_skare3_local_repo(update)
-    all_meta = glob.glob(
-        os.path.join(CONFIG["data_dir"], "skare3", "pkg_defs", "*", "meta.yaml")
-    )
+    all_meta = glob.glob(CONFIG["data_dir"] / "skare3" / "pkg_defs" / "*" / "meta.yaml")
     all_info = []
     for f in all_meta:
         macro = "{% macro compiler(arg) %}{% endmacro %}\n"
         info = yaml.load(
             jinja2.Template(macro + open(f).read()).render(), Loader=yaml.FullLoader
         )
         pkg_info = {
-            "name": os.path.basename(os.path.dirname(f)),
+            "name": f.parent.name,
             "package": info["package"]["name"],
             "repository": None,
             "owner": None,

skare3_tools/scripts/skare3_release_check.py

@@ -25,6 +25,7 @@
 import os
 import re
 import sys
+from pathlib import Path
 
 import git
 import yaml
@@ -38,7 +39,7 @@ def parser():
     parse = argparse.ArgumentParser(description=__doc__)
     parse.add_argument("--version", required=True, help="Target version to build")
     parse.add_argument(
-        "--skare3-path", default=".", help="local copy of the skare3 repo"
+        "--skare3-path", default=".", help="local copy of the skare3 repo", type=Path
     )
     parse.add_argument(
         "--repository", default="sot/skare3", help="Github repository name"
@@ -59,7 +60,7 @@ def parser():
 def main():
     arg_parser = parser()
     args = arg_parser.parse_args()
-    args.skare3_path = os.path.abspath(args.skare3_path)
+    args.skare3_path = args.skare3_path.resolve()
 
     git_repo = None
     try:
@@ -180,7 +181,7 @@ def main():
     logging.info(f"Target version {tag_name}")
     # checking package versions
     # whenever a version equals `branch_name`, replace it by the full version.
-    files = glob.glob(os.path.join(args.skare3_path, "pkg_defs", "ska3-*", "meta.yaml"))
+    files = glob.glob(args.skare3_path / "pkg_defs" / "ska3-*" / "meta.yaml")
     packages = []
     possible_error = []
     for filename in files:

skare3_tools/scripts/test_results.py

@@ -6,12 +6,12 @@
 import argparse
 import importlib
 import json
-import os
 import re
+from pathlib import Path
 
 
 def test_results(directory):
-    filename = os.path.join(directory, "test.log")
+    filename = directory / "test.log"
     with open(filename) as f:
         for line in f:
             if re.match(r"\*\*\*\s+Package\s+Script\s+Status\s+\*\*\*", line):
@@ -35,7 +35,7 @@ def test_results(directory):
             result_dict[k[0]]["version"] = version
 
             test_results = {
-                "log_directory": os.path.basename(directory),
+                "log_directory": directory.name,
                 "results": result_dict,
             }
     return test_results
@@ -47,11 +47,13 @@ def parser():
         "directory",
         help="Directory containing all test results."
         "It must contain a file named test.log",
+        type=Path,
     )
     parse.add_argument(
         "-o",
         help="Output file name (default: test_results.json)",
         default="test_results.json",
+        type=Path,
     )
     return parse