v2.0a10 Release PR

pycentral/base.py

@@ -14,10 +14,9 @@
     save_access_token,
 )
 from .scopes import Scopes
-from .utils.url_utils import NewCentralURLs
+from .utils import AUTHENTICATION
 from .exceptions import LoginError, ResponseError
 
-urls = NewCentralURLs()
 SUPPORTED_API_METHODS = ("POST", "PATCH", "DELETE", "GET", "PUT")
 
 
@@ -100,7 +99,7 @@ def create_token(self, app_name):
         try:
             self.logger.info(f"Attempting to create new token from {app_name}")
             token = oauth.fetch_token(
-                token_url=urls.Authentication["OAUTH"], auth=auth
+                token_url=AUTHENTICATION["OAUTH"], auth=auth
             )
 
             if "access_token" in token:

pycentral/exceptions/generic_op_error.py

@@ -1,7 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from .pycentral_error import PycentralError
+from pycentral.exceptions.pycentral_error import PycentralError
 
 
 class GenericOperationError(PycentralError):

pycentral/exceptions/login_error.py

@@ -1,7 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from .pycentral_error import PycentralError
+from pycentral.exceptions.pycentral_error import PycentralError
 
 
 class LoginError(PycentralError):

pycentral/exceptions/parameter_error.py

@@ -1,7 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from .verification_error import VerificationError
+from pycentral.exceptions.verification_error import VerificationError
 
 
 class ParameterError(VerificationError):

pycentral/exceptions/response_error.py

@@ -1,7 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from .pycentral_error import PycentralError
+from pycentral.exceptions.pycentral_error import PycentralError
 
 
 class ResponseError(PycentralError):

pycentral/exceptions/unsupported_capability_error.py

@@ -1,7 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from .pycentral_error import PycentralError
+from pycentral.exceptions.pycentral_error import PycentralError
 
 
 class UnsupportedCapabilityError(PycentralError):

pycentral/exceptions/verification_error.py

@@ -1,7 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from .pycentral_error import PycentralError
+from pycentral.exceptions.pycentral_error import PycentralError
 
 
 class VerificationError(PycentralError):

pycentral/glp/devices.py

@@ -1,13 +1,11 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from ..utils.url_utils import NewCentralURLs, urlJoin
+from ..utils import GLP_URLS, generate_url
 from .subscriptions import Subscriptions
 from ..utils.glp_utils import check_progress, rate_limit_check
 import time
 
-urls = NewCentralURLs()
-
 DEVICE_GET_LIMIT = 2000
 # Input size per request for DEVICE module APIs.
 INPUT_SIZE = 5
@@ -87,7 +85,7 @@ def get_device(
         """
 
         conn.logger.info("Getting a device in GLP workspace")
-        path = urls.GLP_DEVICES["DEFAULT"]
+        path = generate_url(GLP_URLS["DEVICE"], category="devices")
 
         params = {"limit": limit, "offset": offset}
         if filter:
@@ -142,7 +140,7 @@ def get_status(self, conn, id):
         :rtype: dict
         """
 
-        path = urlJoin(urls.GLP_DEVICES["GET_ASYNC"], id)
+        path = generate_url(f"{GLP_URLS['ASYNC']}/{id}", category="devices")
         resp = conn.command("GET", path, "glp")
         return resp
 
@@ -181,7 +179,7 @@ def add_devices(self, conn, network=[], compute=[], storage=[]):
             resp_list.append(self.__add_dev("storage", storage))
             return resp_list
         else:
-            path = urls.GLP_DEVICES["DEFAULT"]
+            path = generate_url(GLP_URLS["DEVICE"], category="devices")
             data = {"network": network, "compute": compute, "storage": storage}
             resp = conn.command("POST", path, "glp", api_data=data)
             resp_list.append(resp)
@@ -209,7 +207,7 @@ def __add_dev(self, conn, type, inputs):
         :rtype: list
         """
 
-        path = urls.GLP_DEVICES["DEFAULT"]
+        path = generate_url(GLP_URLS["DEVICE"], category="devices")
         data = {"network": [], "compute": [], "storage": []}
 
         if len(inputs) > INPUT_SIZE:
@@ -300,7 +298,7 @@ def add_sub(self, conn, devices, sub, serial=False, key=False):
         # Setup variables for iterating commands.
         queue = [devices] if not split_input else split_input
         resp_list = []
-        path = urls.GLP_DEVICES["DEFAULT"]
+        path = generate_url(GLP_URLS["DEVICE"], category="devices")
         body = {"subscription": [{"id": sub}]}
 
         for inputs in queue:
@@ -372,7 +370,7 @@ def remove_sub(self, conn, devices, serial=False):
         # Setup variables for iterating commands.
         queue = [devices] if not split_input else split_input
         resp_list = []
-        path = urls.GLP_DEVICES["DEFAULT"]
+        path = generate_url(GLP_URLS["DEVICE"], category="devices")
         body = {"subscription": []}
 
         for inputs in queue:
@@ -421,7 +419,7 @@ def assign_devices(
         :rtype: dict
         """
         conn.logger.info("Assigning device(s) to an application")
-        path = urls.GLP_DEVICES["DEFAULT"]
+        path = generate_url(GLP_URLS["DEVICE"], category="devices")
 
         if serial:
             d_list = []
@@ -498,7 +496,7 @@ def unassign_devices(self, conn, devices=None, serial=False):
         :rtype: dict
         """
         conn.logger.info("Unassigning device(s) from an application")
-        path = urls.GLP_DEVICES["DEFAULT"]
+        path = generate_url(GLP_URLS["DEVICE"], category="devices")
 
         if serial:
             d_list = []

pycentral/glp/service_manager.py

@@ -1,9 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from ..utils.url_utils import NewCentralURLs
-
-urls = NewCentralURLs()
+from ..utils import GLP_URLS, generate_url
 
 
 class ServiceManager(object):
@@ -93,7 +91,9 @@ def get_service_manager_provisions(self, conn, limit=2000, offset=0):
         :rtype: dict
         """
         conn.logger.info("Getting provisioned services in GLP workspace")
-        path = urls.GLP_SERVICES["SERVICE_MANAGER_PROVISIONS"]
+        path = generate_url(
+            GLP_URLS["SERVICE_MANAGER_PROVISIONS"], category="service_catalog"
+        )
 
         params = {
             "limit": limit,
@@ -116,7 +116,9 @@ def get_service_manager_by_region(self, conn):
         :rtype: dict
         """
         conn.logger.info("Getting services managers by region in GLP")
-        path = urls.GLP_SERVICES["SERVICE_MANAGER_BY_REGION"]
+        path = generate_url(
+            GLP_URLS["SERVICE_MANAGER_BY_REGION"], category="service_catalog"
+        )
 
         resp = conn.command(api_method="GET", api_path=path, app_name="glp")
         return resp
@@ -161,7 +163,9 @@ def get_service_managers(self, conn, limit=2000, offset=0):
         :rtype: dict
         """
         conn.logger.info("Getting service managers in GLP")
-        path = urls.GLP_SERVICES["SERVICE_MANAGER"]
+        path = generate_url(
+            GLP_URLS["SERVICE_MANAGER"], category="service_catalog"
+        )
 
         params = {
             "limit": limit,

pycentral/glp/subscriptions.py

@@ -1,12 +1,10 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from ..utils.url_utils import NewCentralURLs, urlJoin
+from ..utils import GLP_URLS, generate_url
 from ..utils.glp_utils import rate_limit_check, check_progress
 import time
 
-urls = NewCentralURLs()
-
 # This is the single input size limit for using the POST request endpoint for adding subscriptions.
 INPUT_SIZE = 5
 # This is the rate limit per minute of requests that can be processed by the POST request endpoint for adding subscriptions.
@@ -84,7 +82,7 @@ def get_subscription(
         :return: API response
         :rtype: dict
         """
-        path = urls.GLP_SUBSCRIPTION["DEFAULT"]
+        path = generate_url(GLP_URLS["SUBSCRIPTION"], category="subscriptions")
 
         params = {"limit": limit, "offset": offset}
         if filter:
@@ -134,7 +132,9 @@ def get_status(self, conn, id):
         :rtype: dict
         """
 
-        path = urlJoin(urls.GLP_SUBSCRIPTION["GET_ASYNC"], id)
+        path = generate_url(
+            f"{GLP_URLS["ASYNC"]}/{id}", category="subscriptions"
+        )
         resp = conn.command("GET", path, "glp")
         return resp
 
@@ -154,7 +154,7 @@ def add_subscription(self, conn, subscriptions=None, limit=0, offset=0):
         :return: API response
         :rtype: dict
         """
-        path = urls.GLP_SUBSCRIPTION["DEFAULT"]
+        path = generate_url(GLP_URLS["SUBSCRIPTION"], category="subscriptions")
 
         if len(subscriptions) > INPUT_SIZE:
             resp = []

pycentral/glp/user_management.py

@@ -1,9 +1,7 @@
 # (C) Copyright 2025 Hewlett Packard Enterprise Development LP.
 # MIT License
 
-from ..utils.url_utils import NewCentralURLs, urlJoin
-
-urls = NewCentralURLs()
+from ..utils import GLP_URLS, generate_url
 
 
 class UserMgmt(object):
@@ -39,7 +37,9 @@ def get_users(self, conn, filter=None, limit=300, offset=0):
         :return: API response
         :rtype: dict
         """
-        path = urls.GLP_USER_MANAGEMENT["GET"]
+        path = generate_url(
+            GLP_URLS["USER_MANAGEMENT"], category="user_management"
+        )
 
         params = {
             "limit": limit,
@@ -69,7 +69,9 @@ def get_user(self, conn, email=None, id=None):
         if email:
             id = self.get_user_id(conn, email)[1]
 
-        path = urlJoin(urls.GLP_USER_MANAGEMENT["GET_USER"], id)
+        path = generate_url(
+            f"{GLP_URLS["USER_MANAGEMENT"]}/{id}", category="user_management"
+        )
 
         resp = conn.command("GET", path, "glp")
         if resp["code"] == 200:
@@ -116,7 +118,10 @@ def delete_user(self, conn, email=None, user_id=None):
         if email:
             user_id = self.get_user_id(conn, email)[1]
 
-        path = urlJoin(urls.GLP_USER_MANAGEMENT["DELETE"], user_id)
+        path = generate_url(
+            f"{GLP_URLS["USER_MANAGEMENT"]}/{user_id}",
+            category="user_management",
+        )
         resp = conn.command(api_method="DELETE", api_path=path, app_name="glp")
         return resp
 
@@ -134,11 +139,13 @@ def inv_user(self, conn, email, send_link):
         :rtype: dict
         """
 
-        path = urls.GLP_USER_MANAGEMENT["POST"]
+        path = generate_url(
+            GLP_URLS["USER_MANAGEMENT"], category="user_management"
+        )
         body = {"email": email, "sendWelcomeEmail": send_link}
 
         resp = conn.command("POST", path, "glp", api_data=body)
-        if resp["code"] == 200:
+        if resp["code"] == 201:
             conn.logger.info("Invite user successful!")
         else:
             conn.logger.error("Invite user failed!")

pycentral/new_monitoring/__init__.py

@@ -0,0 +1,5 @@
+from .aps import MonitoringAPs
+from .devices import MonitoringDevices
+from .gateways import MonitoringGateways
+from .sites import MonitoringSites
+from .clients import Clients