Add jsonpatch create and apply

.github/workflows/main.yaml

@@ -90,7 +90,7 @@ jobs:
       matrix:
         python-version: [ '3.8', '3.9', '3.10' ]
         platform: [ubuntu-latest, macOS-latest, windows-latest]
-        extra: [ 'jinja', 'ttp']
+        extra: [ 'jinja', 'ttp', 'jsonpatch']
     runs-on: ${{ matrix.platform }}
     steps:
       - uses: actions/checkout@v2

README.md

@@ -31,6 +31,7 @@ The following groups are available (more details in the pyproject.toml):
 - scrapli
 - nornir
 - pandas
+- jsonpatch
 - tui
 
 ```bash
@@ -134,6 +135,18 @@ Many features are not implemented yet and many features will come.
 
 ![yaml dump](imgs/yaml-dump.png)
 
+
+### JSON Patch ([RFC 6902](http://tools.ietf.org/html/rfc6902))
+
+#### create
+
+![JSON Patch create](imgs/jsonpatch-create.png)
+
+#### apply
+
+![JSON Patch apply](imgs/jsonpatch-apply.png)
+
+
 ### Help
 
 ![Help QRcode](imgs/nettowel-help.png)

nettowel/cli/jsonpatch.py

@@ -0,0 +1,163 @@
+from typing import List
+
+import typer
+from rich import print_json, print
+from rich.panel import Panel
+from rich.columns import Columns
+from rich.json import JSON
+
+from nettowel.cli._common import (
+    auto_complete_paths,
+    read_yaml,
+    get_typer_app,
+)
+from nettowel.exceptions import NettowelInputError
+
+from nettowel import jsonpatch
+
+
+app = get_typer_app(help="JSON Patch [RFC 6902](http://tools.ietf.org/html/rfc6902)")
+
+
+@app.command()
+def create(
+    ctx: typer.Context,
+    src_file_name: typer.FileText = typer.Argument(
+        ...,
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        resolve_path=True,
+        allow_dash=True,
+        metavar="src",
+        help="Source data (YAML/JSON)",
+        autocompletion=auto_complete_paths,
+    ),
+    dst_file_name: typer.FileText = typer.Argument(
+        ...,
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        resolve_path=True,
+        allow_dash=True,
+        metavar="dst",
+        help="Destination data (YAML/JSON)",
+        autocompletion=auto_complete_paths,
+    ),
+    json: bool = typer.Option(False, "--json", help="JSON output"),
+    raw: bool = typer.Option(False, "--raw", help="Raw result output"),
+    only_result: bool = typer.Option(
+        False, "--print-result-only", help="Only print the result"
+    ),
+) -> None:
+    try:
+        src = read_yaml(src_file_name)
+        dst = read_yaml(dst_file_name)
+
+        patch = jsonpatch.create(src=src, dst=dst)
+        patch_str = patch.to_string()
+        if json or raw:
+            print_json(json=patch_str)
+        else:
+            panels: List[Panel] = []
+            if not only_result:
+                panels.append(
+                    Panel(
+                        JSON.from_data(src), title="[yellow]Source", border_style="blue"
+                    )
+                )
+                panels.append(
+                    Panel(
+                        JSON.from_data(dst),
+                        title="[yellow]Destrination",
+                        border_style="blue",
+                    )
+                )
+            panels.append(
+                Panel(JSON(patch_str), title="[yellow]JSON Patch", border_style="blue")
+            )
+            print(Columns(panels, equal=True))
+        raise typer.Exit(0)
+    except NettowelInputError as exc:
+        typer.echo("Input is not valide", err=True)
+        typer.echo(str(exc), err=True)
+        raise typer.Exit(3)
+
+
+@app.command()
+def apply(
+    ctx: typer.Context,
+    patch_file_name: typer.FileText = typer.Argument(
+        ...,
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        resolve_path=True,
+        allow_dash=True,
+        metavar="patch",
+        help="Patch opterations (list of mappings) (YAML/JSON)",
+        autocompletion=auto_complete_paths,
+    ),
+    data_file_name: typer.FileText = typer.Argument(
+        ...,
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        resolve_path=True,
+        allow_dash=True,
+        metavar="data",
+        help="Data to patch (YAML/JSON)",
+        autocompletion=auto_complete_paths,
+    ),
+    json: bool = typer.Option(False, "--json", help="JSON output"),
+    raw: bool = typer.Option(False, "--raw", help="Raw result output"),
+    only_result: bool = typer.Option(
+        False, "--print-result-only", help="Only print the result"
+    ),
+) -> None:
+    try:
+        patch_data = read_yaml(patch_file_name)
+        data_input = read_yaml(data_file_name)
+
+        new_data = jsonpatch.apply(patch=patch_data, data=data_input)
+
+        if json or raw:
+            print_json(data=new_data)
+        else:
+            panels: List[Panel] = []
+            if not only_result:
+                panels.append(
+                    Panel(
+                        JSON.from_data(patch_data),
+                        title="[yellow]Patch",
+                        border_style="blue",
+                    )
+                )
+                panels.append(
+                    Panel(
+                        JSON.from_data(data_input),
+                        title="[yellow]Input Data",
+                        border_style="blue",
+                    )
+                )
+            panels.append(
+                Panel(
+                    JSON.from_data(data=new_data),
+                    title="[yellow]Patched Data with JSON Patch",
+                    border_style="blue",
+                )
+            )
+            print(Columns(panels, equal=True))
+        raise typer.Exit(0)
+    except NettowelInputError as exc:
+        typer.echo("Input is not valide", err=True)
+        typer.echo(str(exc), err=True)
+        raise typer.Exit(3)
+
+
+if __name__ == "__main__":
+    app()

nettowel/cli/main.py

@@ -21,6 +21,7 @@
 from nettowel.cli.scrapli import app as scrapli_app
 from nettowel.cli.restconf import app as restconf_app
 from nettowel.cli.pandas import app as pandas_app
+from nettowel.cli.jsonpatch import app as jsonpatch_app
 from nettowel.cli.help import get_qrcode, HELP_MARKDOWN
 
 from nettowel.exceptions import NettowelDependencyMissing
@@ -40,6 +41,7 @@
     (scrapli_app, "scrapli"),
     (restconf_app, "restconf"),
     (pandas_app, "pandas"),
+    (jsonpatch_app, "jsonpatch"),
 ]:
     app.add_typer(subapp, name=name)
 

nettowel/jsonpatch.py

@@ -0,0 +1,46 @@
+from typing import Any, List, Dict
+from nettowel.logger import log
+from nettowel._common import needs
+
+_module = "jsonpatch"
+
+try:
+    from jsonpatch import JsonPatch
+
+    log.debug("Successfully imported %s", _module)
+    JSONPATCH_INSTALLED = True
+
+except ImportError:
+    log.warning("Failed to import %s", _module)
+    JSONPATCH_INSTALLED = False
+
+
+def create(src: Any, dst: Any) -> "JsonPatch":
+    """Create a JSON patch [RFC 6902](http://tools.ietf.org/html/rfc6902)
+
+    Args:
+        src (any): Data source datastructure
+        dst (any): Data destination datastructure
+
+    Returns:
+        JsonPatch: JsonPatch object containing the patch
+    """
+    needs(JSONPATCH_INSTALLED, "jsonpatch", _module)
+    patch = JsonPatch.from_diff(src=src, dst=dst)
+    return patch
+
+
+def apply(patch: List[Dict[str, Any]], data: Any) -> Any:
+    """Apply a JSON patch [RFC 6902](http://tools.ietf.org/html/rfc6902)
+
+    Args:
+        patch (list[dict]): List of patch instructions
+        data (any): Data to apply the patch onto
+
+    Returns:
+        result: Updated data object
+    """
+    needs(JSONPATCH_INSTALLED, "jsonpatch", _module)
+    jp = JsonPatch(patch)
+    result = jp.apply(data)
+    return result

pyproject.toml

@@ -39,6 +39,8 @@ nornir-netmiko = {version = "^1.0.0", optional = true}
 nornir-rich = {version = "^0.1", optional = true}
 pandas = {version = "^2", optional = true}
 trogon = {version = "^0.5.0", optional = true}
+jsonpatch = {version = "^1.33", optional = true}
+
 
 [tool.poetry.dev-dependencies]
 black = "^22|^23|^24"
@@ -56,6 +58,7 @@ scrapli = ["scrapli"]
 nornir = ["nornir", "nornir-napalm", "nornir-scrapli", "nornir-utils", "nornir-jinja2", "nornir-pyxl", "nornir-http", "nornir-netmiko", "nornir-rich"]
 pandas = ["pandas"]
 tui = ["trogon"]
+jsonpatch = ["jsonpatch"]
 full = [
     "Jinja2", 
     "jinja2schema", 
@@ -74,7 +77,8 @@ full = [
     "nornir-netmiko", 
     "nornir-rich", 
     "pandas",
-    "trogon"
+    "trogon",
+    "jsonpatch"
 ]
 
 [tool.poetry.scripts]

tests/jsonpatch/data1.yaml

@@ -0,0 +1,7 @@
+interfaces:
+  - name: Loopback0
+    addr: 10.10.10.10
+    mask: 255.255.255.255
+  - name: Ethernet0/1
+    addr: 192.168.1.1
+    mask: 255.255.255.0

tests/jsonpatch/data2.yaml

@@ -0,0 +1,7 @@
+interfaces:
+  - name: Loopback0
+    addr: 10.10.10.10
+    mask: 255.255.255.255
+  - name: Ethernet0/1
+    addr: 192.168.1.1
+    mask: 255.255.255.192

tests/jsonpatch/data_patch.json

@@ -0,0 +1,7 @@
+[
+  {
+    "op": "replace",
+    "path": "/interfaces/1/mask",
+    "value": "255.255.255.192"
+  }
+]