Add utils module to check paths in a centralized way (#156)

qgis_deployment_toolbelt/profiles/qdt_profile.py

@@ -27,6 +27,7 @@
 
 # Package
 from qgis_deployment_toolbelt.constants import OS_CONFIG
+from qgis_deployment_toolbelt.utils.check_path import check_path
 
 # #############################################################################
 # ########## Globals ###############
@@ -145,10 +146,19 @@ def from_json(cls, profile_json_path: Path, profile_folder: Path = None) -> Self
 
         """
         # checks
-        if not profile_json_path.is_file():
-            raise FileNotFoundError(f"{profile_json_path.resolve()} does not exist.")
-        if profile_folder and not profile_folder.is_dir():
-            raise TypeError(f"{profile_folder.resolve()} is not a folder.")
+        check_path(
+            input_path=profile_json_path,
+            must_be_a_file=True,
+            must_exists=True,
+            must_be_readable=True,
+        )
+        if profile_folder:
+            check_path(
+                input_path=profile_folder,
+                must_be_a_folder=True,
+                must_exists=True,
+                must_be_readable=True,
+            )
 
         # load JSON
         with profile_json_path.open(mode="r", encoding="utf8") as in_profile_json:

qgis_deployment_toolbelt/scenarios/scenario_reader.py

@@ -15,13 +15,15 @@
 import logging
 from functools import lru_cache
 from io import BufferedIOBase
-from os import R_OK, access
 from pathlib import Path
 from typing import List, Tuple, Union
 
 # 3rd party
 import yaml
 
+# package
+from qgis_deployment_toolbelt.utils.check_path import check_path
+
 # #############################################################################
 # ########## Globals ###############
 # ##################################
@@ -65,25 +67,13 @@ def check_yaml_file(self, yaml_path: Union[str, Path]) -> Path:
         :rtype: Path
         """
         # if path as string load it in Path object
-        if isinstance(yaml_path, str):
-            try:
-                yaml_path = Path(yaml_path)
-            except Exception as exc:
-                raise TypeError("Converting yaml path failed: {}".format(exc))
-
-        # check if file exists
-        if not yaml_path.exists():
-            raise FileExistsError(
-                "YAML file to check doesn't exist: {}".format(yaml_path.resolve())
-            )
-
-        # check if it's a file
-        if not yaml_path.is_file():
-            raise IOError("YAML file is not a file: {}".format(yaml_path.resolve()))
-
-        # check if file is readable
-        if not access(yaml_path, R_OK):
-            raise IOError("yaml file isn't readable: {}".format(yaml_path))
+        check_path(
+            input_path=yaml_path,
+            must_be_a_file=True,
+            must_exists=True,
+            must_be_readable=True,
+        )
+        yaml_path = Path(yaml_path)
 
         # check integrity and structure
         with yaml_path.open(mode="r") as in_yaml_file:

qgis_deployment_toolbelt/utils/__init__.py

@@ -1,5 +1,7 @@
 #! python3  # noqa: E265
 
+"""Common tools and helpers."""
+
 # submodules
 from .proxies import get_proxy_settings  # noqa: E402 F401
 from .str2bool import str2bool  # noqa: E402 F401

qgis_deployment_toolbelt/utils/check_path.py

@@ -0,0 +1,245 @@
+#! python3  # noqa: E265
+
+"""
+    Helpers to check file: readable, exists, etc..
+
+    Author: Julien Moura (https://github.com/guts)
+"""
+
+# #############################################################################
+# ########## Libraries #############
+# ##################################
+
+# Standard library
+import logging
+from os import R_OK, W_OK, access
+from pathlib import Path
+from typing import Union
+
+# #############################################################################
+# ########## Globals ###############
+# ##################################
+
+# logs
+logger = logging.getLogger(__name__)
+
+# #############################################################################
+# ########## Functions #############
+# ##################################
+
+
+def check_var_can_be_path(input_var: str, raise_error: bool = True) -> bool:
+    """Check is the path can be converted as pathlib.Path.
+
+    Args:
+        input_var (str): var to check
+        raise_error (bool, optional): if True, it raises an exception. Defaults to True.
+
+    Raises:
+        TypeError: if input path can't be converted and raise_error is False
+
+    Returns:
+        bool: True if the input can be converted to pathlib.Path
+    """
+    try:
+        input_var = Path(input_var)
+        return True
+    except Exception as exc:
+        error_message = f"Converting {input_var} into Path failed. Trace: {exc}"
+        if raise_error:
+            raise TypeError(error_message)
+        else:
+            logger.error(error_message)
+            return False
+
+
+def check_path_exists(input_path: Union[str, Path], raise_error: bool = True) -> bool:
+    """Check if the input path (file or folder) exists.
+
+    Args:
+        input_path (Union[str, Path]): path to check
+        raise_error (bool, optional): if True, it raises an exception. Defaults to True.
+
+    Raises:
+        FileExistsError: if the path doesn't exist and raise_error is False
+
+    Returns:
+        bool: True if the path exists.
+    """
+    if not isinstance(input_path, Path):
+        if (
+            not check_var_can_be_path(input_path, raise_error=raise_error)
+            and not raise_error
+        ):
+            return False
+        # if previous check passed, let's convert it safely
+        input_path = Path(input_path)
+    if not input_path.exists():
+        error_message = f"{input_path.resolve()} doesn't exist."
+        if raise_error:
+            raise FileExistsError(error_message)
+        else:
+            logger.error(error_message)
+            return False
+    else:
+        return True
+
+
+def check_path_is_readable(input_path: Path, raise_error: bool = True) -> bool:
+    """Check if the input path (file or folder) is readable.
+
+    Args:
+        input_path (Path): path to check
+        raise_error (bool, optional): if True, it raises an exception. Defaults to True.
+
+    Raises:
+        FileExistsError: if the path is not readable and raise_error is False
+
+    Returns:
+        bool: True if the path is readable.
+    """
+    # firstly check the path is valid and exists
+    if not isinstance(input_path, Path):
+        if (
+            not check_path_exists(input_path, raise_error=raise_error)
+            and not raise_error
+        ):
+            return False
+        # if previous check passed, let's convert it safely
+        input_path = Path(input_path)
+
+    if not access(input_path, R_OK):
+        error_message = f"{input_path.resolve()} isn't readable."
+        if raise_error:
+            raise IOError(error_message)
+        else:
+            logger.error(f"{input_path.resolve()} isn't readable.")
+            return False
+    else:
+        return True
+
+
+def check_path_is_writable(input_path: Path, raise_error: bool = True) -> bool:
+    """Check if the input path (file or folder) is writable.
+
+    Args:
+        input_path (Path): path to check
+        raise_error (bool, optional): if True, it raises an exception. Defaults to True.
+
+    Raises:
+        FileExistsError: if the path is not writable and raise_error is False
+
+    Returns:
+        bool: True if the path is writable.
+    """
+    # firstly check the path is valid and exists
+    if not isinstance(input_path, Path):
+        if (
+            not check_path_exists(input_path, raise_error=raise_error)
+            and not raise_error
+        ):
+            return False
+        # if previous check passed, let's convert it safely
+        input_path = Path(input_path)
+
+    if not access(input_path, W_OK):
+        error_message = f"{input_path.resolve()} isn't writable."
+        if raise_error:
+            raise IOError(error_message)
+        else:
+            logger.error(error_message)
+            return False
+    else:
+        return True
+
+
+def check_path(
+    input_path: Union[str, Path],
+    must_exists: bool = True,
+    must_be_readable: bool = True,
+    must_be_writable: bool = False,
+    must_be_a_folder: bool = False,
+    must_be_a_file: bool = False,
+    raise_error: bool = True,
+) -> bool:
+    """Meta function of the module. Check if a given path complies with some constraints.
+
+    Args:
+        input_path (Union[str, Path]): path to check
+        must_exists (bool, optional): path must exist. Defaults to True.
+        must_be_readable (bool, optional): path must be readable. Defaults to True.
+        must_be_writable (bool, optional): path must be writable. Defaults to False.
+        must_be_a_folder (bool, optional): path must be a folder. Mutually exclusive \
+            with must_be_a_file. Defaults to False.
+        must_be_a_file (bool, optional): path must be a file. Mutually exclusive with \
+            must_be_a_folder. Defaults to False.
+        raise_error (bool, optional): if True, it raises an exception. Defaults to True.
+
+    Raises:
+        ValueError: if must_be_a_file and must_be_a_folder are both set to True
+        FileNotFoundError: if the path is not a file and must_be_a_file is set to True
+        NotADirectoryError: if the path is not a folder and must_be_a_folder is set to True
+
+    Returns:
+        bool:  True if the path complies with constraints.
+    """
+    # check arguments
+    if all([must_be_a_file, must_be_a_folder]):
+        raise ValueError(
+            "These options are mutually exclusive: must_be_a_file, must_be_a_folder"
+        )
+
+    # check input path if usable with pathlib.Path
+    if not isinstance(input_path, Path):
+        check_var = check_var_can_be_path(input_var=input_path, raise_error=raise_error)
+        if not check_var and not raise_error:
+            return False
+        input_path = Path(input_path)
+
+    # check
+    if must_exists:
+        check_exist = check_path_exists(input_path=input_path, raise_error=raise_error)
+        if not check_exist and not raise_error:
+            return False
+
+    # check file or folder
+    if must_be_a_file and not input_path.is_file():
+        error_message = f"{input_path.resolve()} is not a file."
+        if raise_error:
+            raise FileNotFoundError(error_message)
+        else:
+            logger.error(error_message)
+            return False
+    if must_be_a_folder and not input_path.is_dir():
+        error_message = f"{input_path.resolve()} is not a folder."
+        if raise_error:
+            raise NotADirectoryError(error_message)
+        else:
+            logger.error(error_message)
+            return False
+
+    # check chmod
+    if must_be_readable:
+        check_readable = check_path_is_readable(
+            input_path=input_path, raise_error=raise_error
+        )
+        if not check_readable and not raise_error:
+            return False
+
+    if must_be_writable:
+        check_writable = check_path_is_writable(
+            input_path=input_path, raise_error=raise_error
+        )
+        if not check_writable and not raise_error:
+            return False
+
+    return True
+
+
+# ############################################################################
+# ##### Stand alone program ########
+# ##################################
+
+if __name__ == "__main__":
+    """Standalone execution."""
+    pass

tests/test_qdt_profile_object.py

@@ -7,7 +7,7 @@
         # for whole tests
         python -m unittest tests.test_qdt_profile_object
         # for specific test
-        python -m unittest tests.test_qdt_profile_object.TestQdtProfile.test_profile_load_from_json
+        python -m unittest tests.test_qdt_profile_object.TestQdtProfile.test_profile_load_from_json_basic
 """
 
 # standard
@@ -36,7 +36,7 @@ def setUpClass(cls):
     def test_profile_load_from_json_basic(self):
         """Test profile loading from JSON."""
         for i in self.good_profiles_files:
-            qdt_profile = QdtProfile.from_json(i)
+            qdt_profile = QdtProfile.from_json(profile_json_path=i)
             self.assertIsInstance(qdt_profile, QdtProfile)
 
             # attributes types