init add Complex Tag functionality

Author: rlabbeptc

kepconfig/adv_tags/__init__.py

@@ -8,7 +8,7 @@
 specific objects within the Kepware Configuration API
 """
 
-from . import adv_tag_group, average_tags, derived_tags
+from . import adv_tag_group, average_tags, derived_tags, complex_tags
 ADV_TAGS_ROOT = '/project/_advancedtags'
 
 def adv_tag_path_split(path: str, *, isItem=False) -> dict:

kepconfig/adv_tags/complex_tags.py

@@ -0,0 +1,138 @@
+# -------------------------------------------------------------------------
+# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
+# See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+r"""`complex_tags` exposes an API to allow modifications (add, delete, modify) to 
+complex tag objects within the Kepware Configuration API
+"""
+
+from ..connection import server
+from ..error import KepError, KepHTTPError
+from ..utils import _url_parse_object
+from typing import Union
+from .. import adv_tags
+
+COMPLEX_TAGS_ROOT = '/complex_tags'
+
+def _get_complex_tags_url(tag: str = None) -> str:
+    '''Creates url object for the "complex_tags" branch of Kepware's project tree.
+    
+    Returns the complex tag specific url when a value is passed as the tag name.
+    '''
+    if tag is None:
+        return COMPLEX_TAGS_ROOT
+    else:
+        return f'{COMPLEX_TAGS_ROOT}/{_url_parse_object(tag)}'
+
+def add_complex_tag(server: server, adv_tag_group_path: str, DATA: Union[dict, list]) -> Union[bool, list]:
+    '''Add `"complex_tag"` or multiple `"complex_tag"` objects to a specific path in Kepware.
+    Can be used to pass a list of complex tags to be added at one path location.
+
+    :param server: instance of the `server` class
+    :param adv_tag_group_path: path identifying where to add complex tag(s). Standard Kepware address decimal 
+    notation string such as "_advancedtags.AdvTagGroup1" or "_advancedtags.AdvTagGroup1.AdvTagGroupChild"
+    :param DATA: Dict or List of Dicts of the complex tag(s) to add
+
+    :return: True - If a "HTTP 201 - Created" is received from Kepware server
+    :return: If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all 
+    complex tags added that failed.
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(adv_tag_group_path, isItem=False)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_complex_tags_url()
+
+    r = server._config_add(url, DATA)
+    if r.code == 201:
+        return True
+    elif r.code == 207:
+        errors = [item for item in r.payload if item['code'] != 201]
+        return errors
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def modify_complex_tag(server: server, complex_tag_path: str, DATA: dict, force: bool = False) -> bool:
+    '''Modify a `"complex_tag"` object and its properties in Kepware.
+
+    :param server: instance of the `server` class
+    :param complex_tag_path: path identifying location and complex tag to modify. Standard Kepware address decimal 
+    notation string including the complex tag such as "_advancedtags.AdvTagGroup1.ComplexTag1"
+    :param DATA: Dict of the `complex_tag` properties to be modified
+    :param force: *(optional)* if True, will force the configuration update to the Kepware server
+
+    :return: True - If a "HTTP 200 - OK" is received from Kepware server
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    complex_tag_data = server._force_update_check(force, DATA)
+    path_obj = adv_tags.adv_tag_path_split(complex_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_complex_tags_url(path_obj['item'])
+
+    r = server._config_update(url, complex_tag_data)
+    if r.code == 200:
+        return True
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def del_complex_tag(server: server, complex_tag_path: str) -> bool:
+    '''Delete `"complex_tag"` object at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param complex_tag_path: path identifying location and complex tag to delete. Standard Kepware address decimal 
+    notation string including the complex tag such as "_advancedtags.AdvTagGroup1.ComplexTag1"
+
+    :return: True - If a "HTTP 200 - OK" is received from Kepware server
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(complex_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_complex_tags_url(path_obj['item'])
+
+    r = server._config_del(url)
+    if r.code == 200:
+        return True
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def get_complex_tag(server: server, complex_tag_path: str) -> dict:
+    '''Returns the properties of the `"complex_tag"` object at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param complex_tag_path: path identifying location and complex tag to retrieve. Standard Kepware address decimal 
+    notation string including the complex tag such as "_advancedtags.AdvTagGroup1.ComplexTag1"
+
+    :return: Dict of data for the complex tag requested
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(complex_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_complex_tags_url(path_obj['item'])
+
+    r = server._config_get(url)
+    return r.payload
+
+def get_all_complex_tags(server: server, adv_tag_group_path: str, *, options: dict = None) -> list:
+    '''Returns the properties of all `"complex_tag"` objects at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param adv_tag_group_path: path identifying location to retrieve complex tag list. Standard Kepware address decimal
+    notation string such as "_advancedtags.AdvTagGroup1" or "_advancedtags.AdvTagGroup1.AdvTagGroupChild"
+    :param options: *(optional)* Dict of parameters to filter, sort or paginate the list of complex tags. Options are `filter`,
+        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
+
+    :return: List of data for all complex tags
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(adv_tag_group_path, isItem=False)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_complex_tags_url()
+
+    r = server._config_get(url, params=options)
+    return r.payload

tests/adv_tags_test.py

@@ -42,6 +42,15 @@
     }
 ]
 
+complex_tag_name = 'ComplexTag1'
+complex_tag_data = [
+    {
+        "common.ALLTYPES_NAME": complex_tag_name,
+        "common.ALLTYPES_DESCRIPTION": "",
+        "advanced_tags.ENABLED": True,
+    }
+]
+
 def HTTPErrorHandler(err):
     if err.__class__ is error.KepHTTPError:
         print(err.code)
@@ -73,6 +82,10 @@ def complete(server: connection.server):
             for obj in obj_list:
                 derived_tag_path = f'_advancedtags.{obj["name"]}'
                 adv_tags.derived_tags.del_derived_tag(server, derived_tag_path)
+        elif key == 'complex_tags':
+            for obj in obj_list:
+                complex_tag_path = f'_advancedtags.{obj["name"]}'
+                adv_tags.complex_tags.del_complex_tag(server, complex_tag_path)
     pass
 
 @pytest.fixture(scope="module")
@@ -205,6 +218,45 @@ def test_derived_tag_del(server):
     derived_tag_path = f'_advancedtags.{adv_tag_group_name}.{derived_tag_name}'
     assert adv_tags.derived_tags.del_derived_tag(server, derived_tag_path)
 
+def test_complex_tag_add(server):
+    # Add a complex tag to the root advanced tag plug-in
+    assert adv_tags.complex_tags.add_complex_tag(server, f'_advancedtags', complex_tag_data)
+
+    testTag = {
+        "common.ALLTYPES_NAME": "newComplexTag",
+        "common.ALLTYPES_DESCRIPTION": "",
+        "advanced_tags.ENABLED": True,
+    }
+    complex_tag_data.append(testTag)
+    # Add a complex tag to the advanced tag group
+    assert adv_tags.complex_tags.add_complex_tag(server, f'_advancedtags.{adv_tag_group_name}', complex_tag_data)
+
+def test_complex_tag_get(server):
+    # Get the complex tag
+    complex_tag_path = f'_advancedtags.{adv_tag_group_name}.{complex_tag_name}'
+    result = adv_tags.complex_tags.get_complex_tag(server, complex_tag_path)
+    assert type(result) == dict
+    assert result.get("common.ALLTYPES_NAME") == complex_tag_name
+
+def test_complex_tag_modify(server):
+    # Modify the complex tag
+    complex_tag_path = f'_advancedtags.{adv_tag_group_name}.{complex_tag_name}'
+    tag_data = {
+        "common.ALLTYPES_DESCRIPTION": "Modified complex tag"
+    }
+    assert adv_tags.complex_tags.modify_complex_tag(server, complex_tag_path, tag_data, force=True)
+
+def test_complex_tag_get_all(server):
+    # Get all complex tags under the group
+    result = adv_tags.complex_tags.get_all_complex_tags(server, f'_advancedtags.{adv_tag_group_name}')
+    assert type(result) == list
+    assert any(tag.get("common.ALLTYPES_NAME") == complex_tag_name for tag in result)
+
+def test_complex_tag_del(server):
+    # Delete the complex tag
+    complex_tag_path = f'_advancedtags.{adv_tag_group_name}.{complex_tag_name}'
+    assert adv_tags.complex_tags.del_complex_tag(server, complex_tag_path)
+
 
 def test_adv_tag_group_del(server):
     # Delete parent advanced tag group