Refactor: Migrate os.path to pathlib.Path in file_processor_handler.py (#68757)

Author: Dotify71

airflow-core/src/airflow/utils/log/file_processor_handler.py

@@ -115,22 +115,22 @@ def _symlink_latest_log_directory(self):
 
         :return: None
         """
-        log_directory = self._get_log_directory()
-        latest_log_directory_path = os.path.join(self.base_log_folder, "latest")
-        if os.path.isdir(log_directory):
-            rel_link_target = Path(log_directory).relative_to(Path(latest_log_directory_path).parent)
+        log_directory = Path(self._get_log_directory())
+        latest_log_directory_path = Path(self.base_log_folder) / "latest"
+        if log_directory.is_dir():
+            rel_link_target = log_directory.relative_to(latest_log_directory_path.parent)
             try:
                 # if symlink exists but is stale, update it
-                if os.path.islink(latest_log_directory_path):
-                    if os.path.realpath(latest_log_directory_path) != log_directory:
-                        os.unlink(latest_log_directory_path)
-                        os.symlink(rel_link_target, latest_log_directory_path)
-                elif os.path.isdir(latest_log_directory_path) or os.path.isfile(latest_log_directory_path):
+                if latest_log_directory_path.is_symlink():
+                    if latest_log_directory_path.resolve() != log_directory.resolve():
+                        latest_log_directory_path.unlink()
+                        latest_log_directory_path.symlink_to(rel_link_target)
+                elif latest_log_directory_path.is_dir() or latest_log_directory_path.is_file():
                     logger.warning(
                         "%s already exists as a dir/file. Skip creating symlink.", latest_log_directory_path
                     )
                 else:
-                    os.symlink(rel_link_target, latest_log_directory_path)
+                    latest_log_directory_path.symlink_to(rel_link_target)
             except OSError:
                 logger.warning("OSError while attempting to symlink the latest log directory")
 
@@ -141,13 +141,13 @@ def _init_file(self, filename):
         :param filename: task instance object
         :return: relative log path of the given task instance
         """
-        relative_log_file_path = os.path.join(self._get_log_directory(), self._render_filename(filename))
-        log_file_path = os.path.abspath(relative_log_file_path)
-        directory = os.path.dirname(log_file_path)
+        relative_log_file_path = Path(self._get_log_directory()) / self._render_filename(filename)
+        log_file_path = relative_log_file_path.resolve()
+        directory = log_file_path.parent
 
-        Path(directory).mkdir(parents=True, exist_ok=True)
+        directory.mkdir(parents=True, exist_ok=True)
 
-        if not os.path.exists(log_file_path):
-            open(log_file_path, "a").close()
+        if not log_file_path.exists():
+            log_file_path.touch()
 
-        return log_file_path
+        return str(log_file_path)