GetJsonPath protocol (#37)

Author: dvolodin7

bi/models/dashboard.py

@@ -1,13 +1,14 @@
 # ----------------------------------------------------------------------
 # Dashboard storage
 # ----------------------------------------------------------------------
-# Copyright (C) 2007-2020 The NOC Project
+# Copyright (C) 2007-2025 The NOC Project
 # See LICENSE for details
 # ----------------------------------------------------------------------
 
 # Python modules
 import datetime
 from base64 import b85encode
+from pathlib import Path
 
 # Third-party modules
 from mongoengine.document import Document, EmbeddedDocument
@@ -27,6 +28,7 @@
 from noc.core.mongo.fields import ForeignKeyField
 from noc.core.prettyjson import to_json
 from noc.core.comp import smart_text
+from noc.core.path import safe_json_path
 
 DAL_NONE = -1
 DAL_RO = 0
@@ -174,5 +176,5 @@ def to_json(self) -> str:
             order=["title", "uuid", "description", "created"],
         )
 
-    def get_json_path(self) -> str:
-        return "%s.json" % self.uuid
+    def get_json_path(self) -> Path:
+        return safe_json_path(str(self.uuid))

bi/models/dashboardlayout.py

@@ -1,16 +1,20 @@
 # ----------------------------------------------------------------------
 # Dashboard Layout
 # ----------------------------------------------------------------------
-# Copyright (C) 2007-2020 The NOC Project
+# Copyright (C) 2007-2025 The NOC Project
 # See LICENSE for details
 # ----------------------------------------------------------------------
 
+# Python modules
+from pathlib import Path
+
 # Third-party modules
 from mongoengine.document import Document, EmbeddedDocument
 from mongoengine.fields import StringField, UUIDField, IntField, ListField, EmbeddedDocumentField
 
 # NOC modules
 from noc.core.prettyjson import to_json
+from noc.core.path import safe_json_path
 
 
 class DashboardCell(EmbeddedDocument):
@@ -72,5 +76,5 @@ def to_json(self) -> str:
             order=["name", "uuid", "description", "cells"],
         )
 
-    def get_json_path(self) -> str:
-        return "%s.json" % self.name
+    def get_json_path(self) -> Path:
+        return safe_json_path(self.name)

cm/models/confdbquery.py

@@ -1,15 +1,15 @@
 # ----------------------------------------------------------------------
 # ConfDBQuery model
 # ----------------------------------------------------------------------
-# Copyright (C) 2007-2021 The NOC Project
+# Copyright (C) 2007-2025 The NOC Project
 # See LICENSE for details
 # ----------------------------------------------------------------------
 
 # Python modules
 import threading
 from typing import Optional, Union
 import operator
-import os
+from pathlib import Path
 
 # Third-party modules
 from bson import ObjectId
@@ -26,9 +26,9 @@
 # NOC modules
 from noc.core.ip import IP
 from noc.core.prettyjson import to_json
-from noc.core.text import quote_safe_path
 from noc.core.model.decorator import on_delete_check
 from noc.sa.interfaces.base import StringParameter, IntParameter, BooleanParameter
+from noc.core.path import safe_json_path
 
 
 class IPParameter(object):
@@ -109,9 +109,8 @@ def __str__(self):
     def get_by_id(cls, oid: Union[str, ObjectId]) -> Optional["ConfDBQuery"]:
         return ConfDBQuery.objects.filter(id=oid).first()
 
-    def get_json_path(self) -> str:
-        p = [quote_safe_path(n.strip()) for n in self.name.split("|")]
-        return os.path.join(*p) + ".json"
+    def get_json_path(self) -> Path:
+        return safe_json_path(self.name)
 
     def query(self, engine, **kwargs):
         """

cm/models/configurationparam.py

@@ -1,17 +1,17 @@
 # ----------------------------------------------------------------------
 # ConfigurationParam model
 # ----------------------------------------------------------------------
-# Copyright (C) 2007-2023 The NOC Project
+# Copyright (C) 2007-2025 The NOC Project
 # See LICENSE for details
 # ----------------------------------------------------------------------
 
 # Python modules
 import threading
 import operator
-import os
 import re
 from dataclasses import dataclass
 from typing import List, Optional, Union, NoReturn, Any
+from pathlib import Path
 
 # Third-party modules
 from bson import ObjectId
@@ -28,9 +28,9 @@
 # NOC modules
 from noc.core.mongo.fields import PlainReferenceField
 from noc.core.prettyjson import to_json
-from noc.core.text import quote_safe_path
 from noc.core.model.decorator import on_delete_check
 from noc.pm.models.metrictype import MetricType
+from noc.core.path import safe_json_path
 from .configurationscope import ConfigurationScope
 
 
@@ -271,9 +271,8 @@ def get_by_id(cls, oid: Union[str, ObjectId]) -> Optional["ConfigurationParam"]:
     def get_by_code(cls, code) -> Optional["ConfigurationParam"]:
         return ConfigurationParam.objects.filter(code=code).first()
 
-    def get_json_path(self) -> str:
-        p = [quote_safe_path(n.strip()) for n in self.name.split("|")]
-        return os.path.join(*p) + ".json"
+    def get_json_path(self) -> Path:
+        return safe_json_path(self.name)
 
     def to_json(self) -> str:
         r = {

cm/models/configurationscope.py

@@ -1,15 +1,15 @@
 # ----------------------------------------------------------------------
 # Configuration Scope model
 # ----------------------------------------------------------------------
-# Copyright (C) 2007-2023 The NOC Project
+# Copyright (C) 2007-2025 The NOC Project
 # See LICENSE for details
 # ----------------------------------------------------------------------
 
 # Python modules
 import threading
 from typing import Optional, Union
 import operator
-import os
+from pathlib import Path
 
 # Third-party modules
 from bson import ObjectId
@@ -22,8 +22,8 @@
 
 # NOC modules
 from noc.core.prettyjson import to_json
-from noc.core.text import quote_safe_path
 from noc.core.model.decorator import on_delete_check
+from noc.core.path import safe_json_path
 
 
 id_lock = threading.Lock()
@@ -63,9 +63,8 @@ def get_by_id(cls, oid: Union[str, ObjectId]) -> Optional["ConfigurationScope"]:
     def get_by_name(cls, name: str) -> Optional["ConfigurationScope"]:
         return ConfigurationScope.objects.filter(name=name).first()
 
-    def get_json_path(self) -> str:
-        p = [quote_safe_path(n.strip()) for n in self.name.split("|")]
-        return os.path.join(*p) + ".json"
+    def get_json_path(self) -> Path:
+        return safe_json_path(self.name)
 
     @property
     def is_common(self) -> bool:

commands/collection.py

@@ -1,13 +1,14 @@
 # ----------------------------------------------------------------------
 # Collections manipulation
 # ----------------------------------------------------------------------
-# Copyright (C) 2007-2019 The NOC Project
+# Copyright (C) 2007-2025 The NOC Project
 # See LICENSE for details
 # ----------------------------------------------------------------------
 
 # Python modules
 import os
 import argparse
+from pathlib import Path
 
 # Third-party modules
 import orjson
@@ -127,20 +128,20 @@ def handle_export(
         if list_collection is not None:
             if list_collection is True:
                 for c in Collection.iter_collections():
-                    print("%s" % c.name, file=self.stdout)
+                    print(c.name, file=self.stdout)
             else:
                 if list_collection not in MODELS:
                     print("Collection not found", file=self.stdout)
                     return
                 objs = MODELS[list_collection].objects.all().order_by("name")
                 for o in objs:
-                    print('uuid:%s name:"%s"' % (o.uuid, o.name), file=self.stdout)
+                    print(f'uuid:{o.uuid} name:"{o.name}"', file=self.stdout)
         else:
             if not export_path or not export_collections:
                 return
-            if not os.path.isdir(export_path):
-                self.die("Path not found: %s" % export_path)
-
+            root = Path(export_path)
+            if not root.is_dir():
+                self.die(f"Path not found: {export_path}")
             for ecname in export_collections:
                 if ecname not in MODELS:
                     print("Collection not found", file=self.stdout)
@@ -152,8 +153,9 @@ def handle_export(
                     kwargs["uuid__in"] = export_model_uuids
                 objs = MODELS[ecname].objects.filter(**kwargs).order_by("name")
                 for o in objs:
-                    path = os.path.join(export_path, ecname, o.get_json_path())
-                    print('export "%s" to %s' % (getattr(o, "name", o), path), file=self.stdout)
+                    path = root / ecname / o.get_json_path()
+                    x_name = getattr(o, "name", o)
+                    print(f'export "{x_name}" to {path}', file=self.stdout)
                     safe_rewrite(path, o.to_json(), mode=0o644)
 
 

core/collection/base.py

@@ -6,7 +6,6 @@
 # ----------------------------------------------------------------------
 
 # Python modules
-import os
 import zlib
 import csv
 import shutil
@@ -479,7 +478,7 @@ def install(cls, data):
         # Format JSON
         json_data = o.to_json()
         # Write
-        path = os.path.join(cls.PREFIX, c.name, o.get_json_path())
+        path = Path(cls.PREFIX, c.name) / o.get_json_path()
         if "uuid" not in data:
             raise ValueError("Invalid JSON: No UUID")
         c.stdout.write("[%s|%s] Installing %s\n" % (c.name, data["uuid"], path))

core/path.py

@@ -0,0 +1,92 @@
+# ----------------------------------------------------------------------
+# Path utilities
+# ----------------------------------------------------------------------
+# Copyright (C) 2007-2025 Gufo Labs
+# See LICENSE for details
+# ----------------------------------------------------------------------
+
+# Python modules
+from pathlib import Path
+from typing import Optional
+import re
+
+# NOC modules
+from .text import cyr_to_lat
+
+_rx_safe_path = re.compile(r"[^a-z0-9\-\+]+", re.IGNORECASE)
+
+
+def _clean_component(s: str) -> str:
+    """
+    Clean path component.
+
+    Args:
+        s: Input string.
+
+    Returns:
+        Cleaned output.
+    """
+    r = cyr_to_lat(s)
+    r = _rx_safe_path.sub(" ", r).strip()
+    return r.replace(" ", "_")
+
+
+def safe_path(*args: str, sep: Optional[str] = None, suffix: Optional[str] = None) -> Path:
+    """
+    Expand path components to safe relative path.
+
+    Suitable for collections. Unreadable or dangerous symbols are replaced
+    with safe ones.
+
+    Examples:
+
+    ``` python
+    safe_path(name, suffix=".json")
+    ```
+
+    ``` python
+    safe_path(name, sep="|", suffix=".json")
+    ```
+
+    ``` python
+    safe_path(vendor, profile, suffix=".json")
+    ```
+
+    Args:
+        args: Path components.
+        sep: Separator, if set, every `args` element is split by `sep`.
+        suffix: If set, append suffix to filename. Suffix doesn't include
+            dot, by default, so i.e. `.json` form must be used, rather than
+            `json`.
+
+    Returns:
+        Formatted path component
+    """
+    if sep:
+        components = []
+        for arg in args:
+            components.extend(arg.split(sep))
+    else:
+        components = list(args)
+    components = [_clean_component(x) for x in components]
+    path = Path(*(x for x in components if x))
+    if suffix:
+        path = path.with_suffix(suffix)
+    return path
+
+
+def safe_json_path(*args: str, sep: Optional[str] = "|", suffix: Optional[str] = ".json") -> Path:
+    """
+    Expand path components to JSON file path.
+
+    Args:
+        args: Path components.
+        sep: Separator, if set, every `args` element is split by `sep`.
+        suffix: If set, append suffix to filename. Suffix doesn't include
+            dot, by default, so i.e. `.json` form must be used, rather than
+            `json`.
+
+    Returns:
+        Formatted path component
+    """
+    return safe_path(*args, sep=sep, suffix=suffix)

core/protocols/__init__.py

@@ -0,0 +1,23 @@
+# ----------------------------------------------------------------------
+# GetJsonPath protocol
+# ----------------------------------------------------------------------
+# Copyright (C) 2007-2025 Gufo Labs
+# See LICENSE for details
+# ----------------------------------------------------------------------
+
+# Python modules
+from pathlib import Path
+from typing import Protocol
+
+
+class ToJson(Protocol):
+    """
+    Saving to collections support.
+
+    Must implement methods:
+    - get_json_path() - relative path in colelction
+    - to_json() - collection content.
+    """
+
+    def get_json_path(self) -> Path: ...
+    def to_json(self) -> str: ...