Use ruamel instead of PyYAML

.github/workflows/main.yaml

@@ -31,8 +31,8 @@ jobs:
 
       # the if the generated files are up to date
       - run: ./run-jsonschema-gentypes
-      - run: jsonschema2md scan_to_paperless/config-schema.json config.md
-      - run: jsonschema2md scan_to_paperless/process.json process.md
+      - run: jsonschema2md scan_to_paperless/config_schema.json config.md
+      - run: jsonschema2md scan_to_paperless/process_schema.json process.md
       - run: git diff --quiet
       - run: git diff
         if: failure()

Pipfile

@@ -4,7 +4,6 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-pyyaml = "==5.4.1"
 numpy = "==1.20.3"
 scipy = "==1.6.3"
 scikit-image = "==0.18.1"

README.md

@@ -49,6 +49,8 @@ $ echo PATH=$PATH:~/venv/bin >> ~/.bashrc
 Create the configuration file on `<home_config>/scan-to-paperless.yaml` (on Linux it's `~/.config/scan-to-paperless.yaml`), with:
 
 ```yaml
+# yaml-language-server: $schema=https://raw.githubusercontent.com/sbrunner/scan-to-paperless/master/scan_to_paperless/config_schema.json
+
 scan_folder: /home/sbrunner/Paperless/scan/
 scanimage_arguments: # Additional argument passed to the scanimage command
     - --device=... # Use `scanimage --list` to get the possible values

jsonschema-gentypes.yaml

@@ -7,7 +7,7 @@ callbacks:
   - - black
   - - isort
 generate:
-  - source: scan_to_paperless/config-schema.json
+  - source: scan_to_paperless/config_schema.json
     destination: scan_to_paperless/config.py
-  - source: scan_to_paperless/process.json
+  - source: scan_to_paperless/process_schema.json
     destination: scan_to_paperless/process_schema.py

scan_to_paperless/__init__.py

@@ -2,7 +2,7 @@
 import sys
 from typing import cast
 
-import yaml
+from ruamel.yaml.main import YAML
 
 if sys.version_info.minor >= 8:
     from scan_to_paperless import config as stp_config
@@ -23,6 +23,8 @@
 
 def get_config() -> stp_config.Configuration:
     if os.path.exists(CONFIG_PATH):
+        yaml = YAML(typ="safe")
+        yaml.default_flow_style = False
         with open(CONFIG_PATH, encoding="utf-8") as config_file:
-            return cast(stp_config.Configuration, yaml.safe_load(config_file.read()))
+            return cast(stp_config.Configuration, yaml.load(config_file.read()))
     return {}

scan_to_paperless/process.py

@@ -17,8 +17,8 @@
 # read, write, rotate, crop, sharpen, draw_line, find_line, find_contour
 import cv2
 import numpy as np
-import yaml
 from deskew import determine_skew_dev
+from ruamel.yaml.main import YAML
 from scipy.signal import find_peaks
 from skimage.color import rgb2gray
 from skimage.metrics import structural_similarity
@@ -133,16 +133,18 @@ def add_intermediate_error(
 
     old_intermediate_error: List[scan_to_paperless.process_schema.IntermediateError] = []
     old_intermediate_error.extend(config["intermediate_error"])
+    yaml = YAML(typ="safe")
+    yaml.default_flow_style = False
     try:
         config["intermediate_error"].append({"error": str(error), "traceback": traceback_})
         with open(config_file_name + "_", "w") as config_file:
-            config_file.write(yaml.safe_dump(config, default_flow_style=False))
+            yaml.dump(config, config_file)
     except Exception as exception:
         print(exception)
         config["intermediate_error"] = old_intermediate_error
         config["intermediate_error"].append({"error": str(error), "traceback": traceback_})
         with open(config_file_name + "_", "w") as config_file:
-            config_file.write(yaml.safe_dump(config, default_flow_style=False))
+            yaml.dump(config, config_file)
     os.rename(config_file_name + "_", config_file_name)
 
 
@@ -988,8 +990,10 @@ def finalise(
 
 def write_error(root_folder: str, message: str) -> None:
     if not os.path.exists(os.path.join(root_folder, "error.yaml")):
+        yaml = YAML(typ="safe")
+        yaml.default_flow_style = False
         with open(os.path.join(root_folder, "error.yaml"), "w") as error_file:
-            error_file.write(yaml.safe_dump({"error": message}, default_flow_style=False))
+            yaml.dump({"error": message}, error_file)
 
 
 def is_sources_present(step: scan_to_paperless.process_schema.Step, root_folder: str) -> bool:
@@ -1000,8 +1004,10 @@ def is_sources_present(step: scan_to_paperless.process_schema.Step, root_folder:
 
 
 def save_config(config: scan_to_paperless.process_schema.Configuration, config_file_name: str) -> None:
+    yaml = YAML(typ="safe")
+    yaml.default_flow_style = False
     with open(config_file_name + "_", "w") as config_file:
-        config_file.write(yaml.safe_dump(config, default_flow_style=False))
+        yaml.dump(config, config_file)
     os.rename(config_file_name + "_", config_file_name)
 
 
@@ -1024,8 +1030,10 @@ def main() -> None:
             if os.path.exists(os.path.join(root_folder, "error.yaml")):
                 continue
 
+            yaml = YAML(typ="safe")
+            yaml.default_flow_style = False
             with open(config_file_name) as config_file:
-                config: scan_to_paperless.process_schema.Configuration = yaml.safe_load(config_file.read())
+                config: scan_to_paperless.process_schema.Configuration = yaml.load(config_file.read())
             if config is None:
                 write_error(root_folder, "Empty config")
                 continue
@@ -1082,22 +1090,20 @@ def main() -> None:
                     continue
             except Exception as exception:
                 print(exception)
+                yaml = YAML(typ="safe")
+                yaml.default_flow_style = False
                 try:
                     with open(os.path.join(root_folder, "error.yaml"), "w") as error_file:
-                        error_file.write(
-                            yaml.dump(
-                                {"error": exception, "traceback": traceback.format_exc().split("\n")},
-                                default_flow_style=False,
-                            )
+                        yaml.dump(
+                            {"error": exception, "traceback": traceback.format_exc().split("\n")},
+                            error_file,
                         )
                 except Exception as exception2:
                     print(exception2)
                     with open(os.path.join(root_folder, "error.yaml"), "w") as error_file:
-                        error_file.write(
-                            yaml.safe_dump(
-                                {"error": str(exception2), "traceback": traceback.format_exc().split("\n")},
-                                default_flow_style=False,
-                            )
+                        yaml.dump(
+                            {"error": str(exception2), "traceback": traceback.format_exc().split("\n")},
+                            error_file,
                         )
 
         sys.stdout.flush()

scan_to_paperless/scan.py

@@ -97,7 +97,7 @@ def add_argument(name: str, choices: Optional[List[str]] = None, **kwargs: Any)
         yaml = YAML(typ="safe")
         yaml.default_flow_style = False
         with open(CONFIG_PATH, "w", encoding="utf-8") as config_file:
-            config_file.write(yaml.dump(config))
+            yaml.dump(config, config_file)
 
     if "scan_folder" not in config:
         print(
@@ -195,9 +195,9 @@ def image_match(image_name: str) -> int:
         with open(os.path.join(os.path.dirname(root_folder), "config.yaml"), "w") as config_file:
             config_file.write(
                 "# yaml-language-server: $schema=https://raw.githubusercontent.com/sbrunner/scan-to-paperless"
-                "/master/schema.json\n\n"
+                "/master/scan_to_paperless/process_schema.json\n\n"
             )
-            config_file.write(yaml.dump(process_config))
+            yaml.dump(process_config, config_file)
 
     except subprocess.CalledProcessError as exception:
         print(exception)

scan_to_paperless/scan_process_status.py

@@ -5,7 +5,7 @@
 import re
 import subprocess  # nosec
 
-import yaml
+from ruamel.yaml.main import YAML
 
 import scan_to_paperless.process_schema
 from scan_to_paperless import get_config
@@ -20,14 +20,14 @@ def main() -> None:
         if not os.path.exists(os.path.join(folder, "config.yaml")):
             print("No config")
         else:
+            yaml = YAML(typ="safe")
+            yaml.default_flow_style = False
             with open(os.path.join(folder, "config.yaml")) as config_file:
-                job_config: scan_to_paperless.process_schema.Configuration = yaml.safe_load(
-                    config_file.read()
-                )
+                job_config: scan_to_paperless.process_schema.Configuration = yaml.load(config_file.read())
 
             if os.path.exists(os.path.join(folder, "error.yaml")):
                 with open(os.path.join(folder, "error.yaml")) as error_file:
-                    error = yaml.safe_load(error_file.read())
+                    error = yaml.load(error_file.read())
                     if error is not None and "error" in error:
                         print(error["error"])
                         if isinstance(error["error"], subprocess.CalledProcessError):

setup.py

@@ -30,7 +30,7 @@
     author_email="stephane.brunner@gmail.com",
     url="https://hub.docker.com/r/sbrunner/scan-to-paperless/",
     packages=find_packages(exclude=["tests.*"]),
-    install_requires=["argcomplete", "pyyaml", "scikit-image"],
+    install_requires=["argcomplete", "ruamel.yaml", "scikit-image"],
     entry_points={
         "console_scripts": [
             "scan = scan_to_paperless.scan:main",