Fixes for #26 and #23

Author: rlabbeptc

kepconfig/__init__.py

@@ -20,5 +20,4 @@
 """
 __version__ = "1.2.0"
 from . import connection, error
-from .utils import path_split
 

kepconfig/admin/ua_server.py

@@ -10,6 +10,7 @@
 from typing import Union
 from ..error import KepHTTPError, KepError
 from ..connection import server
+from ..utils import _url_parse_object
 
 
 UA_ROOT = '/admin/ua_endpoints'
@@ -24,7 +25,7 @@ def _create_url(endpoint = None):
     if endpoint == None:
         return UA_ROOT
     else:
-        return '{}/{}'.format(UA_ROOT,endpoint)
+        return '{}/{}'.format(UA_ROOT, _url_parse_object(endpoint))
 
 def add_endpoint(server: server, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add an `"endpoint"` or multiple `"endpoint"` objects to Kepware UA Server by passing a 

kepconfig/admin/user_groups.py

@@ -10,6 +10,7 @@
 from typing import Union
 from ..error import KepHTTPError, KepError
 from ..connection import server
+from ..utils import _url_parse_object
 
 
 USERGROUPS_ROOT = '/admin/server_usergroups'
@@ -25,7 +26,7 @@ def _create_url(user_group = None):
     if user_group == None:
         return USERGROUPS_ROOT
     else:
-        return '{}/{}'.format(USERGROUPS_ROOT,user_group)
+        return '{}/{}'.format(USERGROUPS_ROOT, _url_parse_object(user_group))
 
 def add_user_group(server: server, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"user group"` or multiple `"user group"` objects to Kepware User Manager by passing a 

kepconfig/admin/users.py

@@ -10,6 +10,7 @@
 from typing import Union
 from ..error import KepError, KepHTTPError
 from ..connection import server
+from ..utils import _url_parse_object
 
 
 USERS_ROOT = '/admin/server_users'
@@ -25,7 +26,7 @@ def _create_url(user = None):
     if user == None:
         return USERS_ROOT
     else:
-        return '{}/{}'.format(USERS_ROOT,user)
+        return '{}/{}'.format(USERS_ROOT, _url_parse_object(user))
 
 def add_user(server: server, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"user"` or multiple `"user"` objects to Kepware User Manager by passing a 

kepconfig/connection.py

@@ -431,7 +431,9 @@ def __connect(self,request_obj):
     # Fucntion used to ensure special characters are handled in the URL
     # Ex: Space will be turned to %20
     def __url_validate(self, url):
-        parsed_url = parse.urlparse(url)
+        # Configuration API does not use fragments in URL so ignore to allow # as a character
+        # Objects in Kepware can include # as part of the object names
+        parsed_url = parse.urlparse(url, allow_fragments= False)
         # Added % for scenarios where special characters have already been escaped with %
         updated_path = parse.quote(parsed_url.path, safe = '/%')
 

kepconfig/connectivity/channel.py

@@ -13,6 +13,7 @@
 import inspect
 from ..connection import server
 from ..error import KepHTTPError, KepError
+from ..utils import _url_parse_object
 from typing import Union
 from . import device
 
@@ -28,7 +29,7 @@ def _create_url(channel = None):
     if channel == None:
         return CHANNEL_ROOT
     else:
-        return '{}/{}'.format(CHANNEL_ROOT,channel)
+        return '{}/{}'.format(CHANNEL_ROOT,_url_parse_object(channel))
 
 def add_channel(server: server, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"channel"` or multiple `"channel"` objects to Kepware. Can be used to pass children of a channel object 
@@ -84,7 +85,7 @@ def modify_channel(server: server, DATA: dict, *, channel: str = None, force: bo
 
     :param server: instance of the `server` class
     :param DATA: Dict of the `channel` properties to be modified
-    :param channel: *(optional)* - name of channel to modify. Only needed if not existing in `"DATA"`
+    :param channel: *(optional)* name of channel to modify. Only needed if not existing in `"DATA"`
     :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

kepconfig/connectivity/device.py

@@ -11,8 +11,8 @@
 
 from ..connection import KepServiceResponse, server
 from ..error import KepHTTPError, KepError
+from ..utils import _url_parse_object, path_split
 from typing import Union
-import kepconfig
 from . import channel, tag
 import inspect
 
@@ -28,7 +28,7 @@ def _create_url(device = None):
     if device == None:
         return DEVICE_ROOT
     else:
-        return '{}/{}'.format(DEVICE_ROOT,device)
+        return '{}/{}'.format(DEVICE_ROOT,_url_parse_object(device))
 
 def add_device(server: server, channel_name: str, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"device"` or multiple `"device"` objects to a channel in Kepware. Can be used to pass children of a device object 
@@ -74,7 +74,7 @@ def del_device(server: server, device_path: str) -> bool:
     :raises KepURLError: If urllib provides an URLError
     '''
 
-    path_obj = kepconfig.path_split(device_path)
+    path_obj = path_split(device_path)
     try:
         r = server._config_del(server.url + channel._create_url(path_obj['channel']) + _create_url(path_obj['device']))
         if r.code == 200: return True 
@@ -100,7 +100,7 @@ def modify_device(server: server, device_path: str, DATA: dict, *, force: bool =
 
     device_data = server._force_update_check(force, DATA)
 
-    path_obj = kepconfig.path_split(device_path)
+    path_obj = path_split(device_path)
     try:
         r = server._config_update(server.url + channel._create_url(path_obj['channel']) + _create_url(path_obj['device']), device_data)
         if r.code == 200: return True 
@@ -122,7 +122,7 @@ def get_device(server: server, device_path: str) -> dict:
     :raises KepURLError: If urllib provides an URLError
     '''
 
-    path_obj = kepconfig.path_split(device_path)
+    path_obj = path_split(device_path)
     try:
         r = server._config_get(server.url + channel._create_url(path_obj['channel']) + _create_url(path_obj['device']))
         return r.payload
@@ -164,7 +164,7 @@ def auto_tag_gen(server: server, device_path: str, job_ttl: int = None) -> KepSe
     :raises KepURLError: If urllib provides an URLError
     '''
 
-    path_obj = kepconfig.path_split(device_path)
+    path_obj = path_split(device_path)
     try:
         url = server.url +channel._create_url(path_obj['channel']) + _create_url(path_obj['device']) + ATG_URL
         job = server._kep_service_execute(url, None, job_ttl)

kepconfig/connectivity/egd/exchange.py

@@ -9,9 +9,9 @@
 exchange objects for EGD devices within the Kepware Configuration API
 """
 
-from ... import path_split
 from ...connection import server
 from ...error import KepHTTPError, KepError
+from ...utils import _url_parse_object, path_split
 from typing import Union
 from .. import egd as EGD, channel, device
 
@@ -34,9 +34,9 @@ def _create_url(device_path, ex_type, exchange_name = None):
             return device_root + PRODUCER_ROOT
     else:
         if ex_type.upper() == EGD.CONSUMER_EXCHANGE:
-            return '{}{}/{}'.format(device_root,CONSUMER_ROOT,exchange_name)
+            return '{}{}/{}'.format(device_root,CONSUMER_ROOT,_url_parse_object(exchange_name))
         else:
-            return '{}{}/{}'.format(device_root,PRODUCER_ROOT,exchange_name)
+            return '{}{}/{}'.format(device_root,PRODUCER_ROOT,_url_parse_object(exchange_name))
 
 def add_exchange(server: server, device_path: str, ex_type: str, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"exchange"` or multiple `"exchange"` objects to Kepware. Can be used to pass children of a exchange object 

kepconfig/connectivity/egd/name.py

@@ -9,7 +9,7 @@
 name resolution objects for EGD devices within the Kepware Configuration API
 """
 
-from ... import path_split
+from ...utils import _url_parse_object, path_split
 from ...connection import server
 from ...error import KepHTTPError, KepError
 from typing import Union
@@ -29,7 +29,7 @@ def _create_url(device_path, name = None):
     if name == None:
         return '{}/{}'.format(device_root, NAMES_ROOT)
     else:
-        return '{}/{}/{}'.format(device_root, NAMES_ROOT, name)
+        return '{}/{}/{}'.format(device_root, NAMES_ROOT, _url_parse_object(name))
 
 def add_name_resolution(server: server, device_path: str, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"name resolution"` or multiple `"name resolution"` objects to Kepware. This allows you to 

kepconfig/connectivity/egd/range.py

@@ -13,6 +13,7 @@
 from .. import egd as EGD
 from ...connection import server
 from ...error import KepError, KepHTTPError
+from ...utils import _url_parse_object
 
 RANGES_ROOT = '/ranges'
 
@@ -27,7 +28,7 @@ def _create_url(device_path, ex_type, exchange_name, range = None):
     if range == None:
         return '{}{}'.format(exchange_root, RANGES_ROOT)
     else:
-        return '{}{}/{}'.format(exchange_root, RANGES_ROOT, range)
+        return '{}{}/{}'.format(exchange_root, RANGES_ROOT, _url_parse_object(range))
 
 def add_range(server: server, device_path: str, ex_type: str, exchange_name: str, DATA: Union[dict, list]) -> Union[bool, list]:
     '''Add a `"range"` or multiple `"range"` objects to Kepware. This allows you to