Replace more os.path with pathlib

pyproject.toml

@@ -125,7 +125,7 @@ line-length = 88
 exclude = ["doc/", "pytensor/_version.py"]
 
 [tool.ruff.lint]
-select = ["C", "E", "F", "I", "UP", "W", "RUF", "PERF"]
+select = ["C", "E", "F", "I", "UP", "W", "RUF", "PERF", "PTH"]
 ignore = ["C408", "C901", "E501", "E741", "RUF012", "PERF203"]
 
 
@@ -136,6 +136,7 @@ lines-after-imports = 2
 # TODO: Get rid of these:
 "**/__init__.py" = ["F401", "E402", "F403"]
 "pytensor/tensor/linalg.py" = ["F403"]
+"pytensor/link/c/cmodule.py" = ["PTH"]
 # For the tests we skip because `pytest.importorskip` is used:
 "tests/link/jax/test_scalar.py" = ["E402"]
 "tests/link/jax/test_tensor_basic.py" = ["E402"]

pytensor/link/c/lazylinker_c.py

-import errno
 import logging
-import os
 import sys
 import warnings
 from importlib import reload
@@ -43,23 +41,12 @@ try:
     # compile_str()) but if another lazylinker_ext does exist then it will be
     # imported and compile_str won't get called at all.
     location = config.compiledir / "lazylinker_ext"
-    if not location.exists():
-        try:
-            # Try to make the location
-            os.mkdir(location)
-        except OSError as e:
-            # If we get an error, verify that the error was # 17, the
-            # path already exists, and that it is a directory Note: we
-            # can't check if it exists before making it, because we
-            # are not holding the lock right now, so we could race
-            # another process and get error 17 if we lose the race
-            assert e.errno == errno.EEXIST
-            assert location.is_dir()
+    location.mkdir(exist_ok=True)
 
     init_file = location / "__init__.py"
     if not init_file.exists():
         try:
-            with open(init_file, "w"):
+            with init_file.open("w"):
                 pass
         except OSError as e:
             if init_file.exists():
@@ -129,12 +116,7 @@ except ImportError:
             code = cfile.read_text("utf-8")
 
             loc = config.compiledir / dirname
-            if not loc.exists():
-                try:
-                    os.mkdir(loc)
-                except OSError as e:
-                    assert e.errno == errno.EEXIST
-                    assert loc.exists()
+            loc.mkdir(exist_ok=True)
 
             args = GCC_compiler.compile_args()
             GCC_compiler.compile_str(dirname, code, location=loc, preargs=args)
@@ -147,8 +129,7 @@ except ImportError:
             # imported at the same time: we need to make sure we do not
             # reload the now outdated __init__.pyc below.
             init_pyc = loc / "__init__.pyc"
-            if init_pyc.is_file():
-                os.remove(init_pyc)
+            init_pyc.unlink(missing_ok=True)
 
             try_import()
             try_reload()

pytensor/utils.py

@@ -156,7 +156,7 @@ def subprocess_Popen(command: str | list[str], **params):
     # with the default None values.
     stdin = None
     if "stdin" not in params:
-        stdin = open(os.devnull)
+        stdin = Path(os.devnull).open()
         params["stdin"] = stdin.fileno()
 
     try:

tests/misc/test_pkl_utils.py

 import os
 import shutil
+from pathlib import Path
 from tempfile import mkdtemp
 
 from pytensor.misc.pkl_utils import StripPickler
 from pytensor.tensor.type import matrix
 
 
+# FIXME: this test looks weird
 class TestStripPickler:
     def setup_method(self):
         # Work in a temporary directory to avoid cluttering the repository
-        self.origdir = os.getcwd()
+        self.origdir = Path.cwd()
         self.tmpdir = mkdtemp()
         os.chdir(self.tmpdir)
 
@@ -20,9 +22,9 @@ class TestStripPickler:
             shutil.rmtree(self.tmpdir)
 
     def test_basic(self):
-        with open("test.pkl", "wb") as f:
+        with Path("test.pkl").open("wb"):
             m = matrix()
             dest_pkl = "my_test.pkl"
-            with open(dest_pkl, "wb") as f:
+            with Path(dest_pkl).open("wb") as f:
                 strip_pickler = StripPickler(f, protocol=-1)
                 strip_pickler.dump(m)

tests/scan/test_basic.py

@@ -13,6 +13,7 @@ import os
 import pickle
 import shutil
 import sys
+from pathlib import Path
 from tempfile import mkdtemp
 
 import numpy as np
@@ -327,15 +328,15 @@ class TestScan:
             [state, n_steps], output, updates=updates, allow_input_downcast=True
         )
 
-        origdir = os.getcwd()
+        origdir = Path.cwd()
         tmpdir = None
         try:
             tmpdir = mkdtemp()
             os.chdir(tmpdir)
 
-            with open("tmp_scan_test_pickle.pkl", "wb") as f_out:
+            with Path("tmp_scan_test_pickle.pkl").open("wb") as f_out:
                 pickle.dump(_my_f, f_out, protocol=-1)
-            with open("tmp_scan_test_pickle.pkl", "rb") as f_in:
+            with Path("tmp_scan_test_pickle.pkl").open("rb") as f_in:
                 my_f = pickle.load(f_in)
         finally:
             # Get back to the original dir, and delete the temporary one.

tests/tensor/utils.py

 import os
 from copy import copy
 from itertools import combinations
+from pathlib import Path
 from tempfile import mkstemp
 
 import numpy as np
@@ -442,7 +443,7 @@ def makeTester(
             gc.collect()
             for f, fname in self.tmp_files:
                 os.close(f)
-                os.remove(fname)
+                Path(fname).unlink()
 
         @pytest.mark.skipif(skip, reason="Skipped")
         def test_good(self):