GLP

The GLP module provides interfaces for device, subscription, and user management through the HPE GreenLake Platform

---

Devices

devices

Devices

Bases: object

get_all_devices(conn, select=None)

Get a list of devices managed in a workspace.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
select	list	A comma separated list of select properties to display in the response. The default is that all properties are returned. Example: select=serialNumber,macAddress	None

Returns:

Type	Description
list[dict]	A list of all devices in the workspace, or an empty list if an error occurs

Source code in pycentral/glp/devices.py

def get_all_devices(self, conn, select=None):
    """Get a list of devices managed in a workspace.

    Args:
        conn (NewCentralBase): pycentral base connection object
        select (list, optional): A comma separated list of select properties to display in
            the response. The default is that all properties are returned.
            Example: select=serialNumber,macAddress

    Returns:
        (list[dict]): A list of all devices in the workspace, or an empty list if an error occurs
    """
    conn.logger.info("Getting all devices in GLP workspace")
    limit = DEVICE_GET_LIMIT
    offset = 0
    device_list = []

    while True:
        resp = self.get_device(
            conn, limit=limit, offset=offset, select=select
        )
        if resp["code"] != 200:
            conn.logger.error(
                f"Error fetching list of devices: {resp['code']} - {resp['msg']}"
            )
            device_list = []
            break
        device_resp_message = resp["msg"]
        device_list.extend(device_resp_message["items"])
        if len(device_list) == device_resp_message["total"]:
            conn.logger.info(
                f"Total devices fetched from account: {len(device_list)}"
            )
            break
        offset += DEVICE_GET_LIMIT
    return device_list

get_device(conn, limit=DEVICE_GET_LIMIT, offset=0, filter=None, select=None, sort=None)

Get a list of devices managed in a GLP workspace from filter/select and sort inputs.

Rate limits are enforced on this API. 160 requests per minute is supported per workspace. API will result in 429 if this threshold is breached.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
limit	int	Specifies the number of results to be returned. The default value is 2000	DEVICE_GET_LIMIT
offset	int	Specifies the zero-based resource offset to start the response from. The default value is 0	0
filter	str	Device filters joined by logical operators	None
select	list	Properties of devices to be displayed in response	None
sort	str	Sort string expressions	None

Returns:

Type	Description
dict	Response as provided by 'command' function in NewCentralBase

Source code in pycentral/glp/devices.py

def get_device(
    self,
    conn,
    limit=DEVICE_GET_LIMIT,
    offset=0,
    filter=None,
    select=None,
    sort=None,
):
    """Get a list of devices managed in a GLP workspace from filter/select and sort inputs.

    Rate limits are enforced on this API. 160 requests per minute is supported per workspace.
    API will result in 429 if this threshold is breached.

    Args:
        conn (NewCentralBase): pycentral base connection object
        limit (int, optional): Specifies the number of results to be returned.
            The default value is 2000
        offset (int, optional): Specifies the zero-based resource offset to start the
            response from. The default value is 0
        filter (str, optional): Device filters joined by logical operators
        select (list, optional): Properties of devices to be displayed in response
        sort (str, optional): Sort string expressions

    Returns:
        (dict): Response as provided by 'command' function in NewCentralBase
    """

    conn.logger.info("Getting a device in GLP workspace")
    path = generate_url(GLP_URLS["DEVICE"], category="devices")

    params = {"limit": limit, "offset": offset}
    if filter:
        params["filter"] = filter
    if select:
        params["select"] = select
    if sort:
        params["sort"] = sort

    resp = conn.command("GET", path, "glp", api_params=params)
    if resp["code"] == 200:
        conn.logger.info("Get device successful!")
    else:
        conn.logger.error("Get device failed!")
    return resp

get_device_id(conn, serial)

Get device ID in a GLP workspace based on serial number.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
serial	str	Device serial number	required

Returns:

Type	Description
tuple(bool, str)	Tuple of two elements. First element returns True if device id is found, else False. The second element is a GLP device ID if found, else an error message from the response

Source code in pycentral/glp/devices.py

def get_device_id(self, conn, serial):
    """Get device ID in a GLP workspace based on serial number.

    Args:
        conn (NewCentralBase): pycentral base connection object
        serial (str): Device serial number

    Returns:
        (tuple(bool, str)): Tuple of two elements. First element returns True if device id
            is found, else False. The second element is a GLP device ID if found, else an
            error message from the response
    """

    filter = f"serialNumber eq '{serial}'"
    resp = self.get_device(conn, filter=filter)
    if resp["code"] != 200:
        return (resp, (False, "Bad request for get_id"))
    elif resp["msg"]["count"] == 0:
        return (False, "Serial not found")
    else:
        return (True, resp["msg"]["items"][0]["id"])

get_status(conn, id)

Get status of an async GLP devices request.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
id	str	Transaction ID from async API request	required

Returns:

Type	Description
dict	Response as provided by 'command' function in NewCentralBase

Source code in pycentral/glp/devices.py

def get_status(self, conn, id):
    """Get status of an async GLP devices request.

    Args:
        conn (NewCentralBase): pycentral base connection object
        id (str): Transaction ID from async API request

    Returns:
        (dict): Response as provided by 'command' function in NewCentralBase
    """

    path = generate_url(f"{GLP_URLS['ASYNC']}/{id}", category="devices")
    resp = conn.command("GET", path, "glp")
    return resp

add_devices(conn, network=[], compute=[], storage=[])

Add devices to a workspace in GreenLake Platform (GLP).

Handles coordinating chaining requests if passed more than 5 devices (max per api call). Can use any combination of network, compute, and storage devices. Always returns a 202 response code if basic input validation is met. Currently does not support Async get status handling to confirm operation success.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
network	list	Network devices as dict objects	[]
compute	list	Compute devices as dict objects	[]
storage	list	Storage devices as dict objects	[]

Returns:

Type	Description
list[dict]	List of response objects as provided by 'command' function in NewCentralBase

Source code in pycentral/glp/devices.py

def add_devices(self, conn, network=[], compute=[], storage=[]):
    """Add devices to a workspace in GreenLake Platform (GLP).

    Handles coordinating chaining requests if passed more than 5 devices (max per api call).
    Can use any combination of network, compute, and storage devices. Always returns a 202
    response code if basic input validation is met. Currently does not support Async get
    status handling to confirm operation success.

    Args:
        conn (NewCentralBase): pycentral base connection object
        network (list, optional): Network devices as dict objects
        compute (list, optional): Compute devices as dict objects
        storage (list, optional): Storage devices as dict objects

    Returns:
        (list[dict]): List of response objects as provided by 'command' function in NewCentralBase
    """

    count = len(network) + len(compute) + len(storage)
    resp_list = []

    # Check for rate limit handler
    if count > INPUT_SIZE:
        conn.logger.info("WARNING MORE THAN 5 DEVICES IS AN ALPHA FEATURE!")
        resp_list.append(self.__add_dev("network", network))
        resp_list.append(self.__add_dev("compute", compute))
        resp_list.append(self.__add_dev("storage", storage))
        return resp_list
    else:
        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)
        if resp["code"] == 202:
            conn.logger.info("Add device request accepted...")
        else:
            conn.logger.error("Add device request failed!")
        return resp_list

__add_dev(conn, type, inputs)

Helper function for add_devices.

Handles splitting inputs larger than input size and coordinates running the commands to not exceed rate limit.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
type	str	One of network, compute, or storage	required
inputs	list	List of 'type' objects in dict format	required

Returns:

Type	Description
list[dict]	Response object(s) as provided by 'command' function in NewCentralBase

Source code in pycentral/glp/devices.py

def __add_dev(self, conn, type, inputs):
    """Helper function for add_devices.

    Handles splitting inputs larger than input size and coordinates running the commands
    to not exceed rate limit.

    Args:
        conn (NewCentralBase): pycentral base connection object
        type (str): One of network, compute, or storage
        inputs (list): List of 'type' objects in dict format

    Returns:
        (list[dict]): Response object(s) as provided by 'command' function in NewCentralBase
    """

    path = generate_url(GLP_URLS["DEVICE"], category="devices")
    data = {"network": [], "compute": [], "storage": []}

    if len(inputs) > INPUT_SIZE:
        split_input, wait_time = rate_limit_check(
            inputs, INPUT_SIZE, POST_RPM
        )

        resp_list = []

        for devices in split_input:
            data["network"] = devices if type == "network" else []
            data["compute"] = devices if type == "compute" else []
            data["storage"] = devices if type == "storage" else []
            resp = conn.command("POST", path, "glp", api_data=data)
            if resp["code"] != 202:
                conn.logger.error(
                    f"Add device request failed for {inputs}!"
                )
            else:
                conn.logger.info("Add device request accepted...")
            resp_list.append(resp)
            time.sleep(wait_time)
        return resp_list
    else:
        data["network"] = inputs if type == "network" else []
        data["compute"] = inputs if type == "compute" else []
        data["storage"] = inputs if type == "storage" else []

        resp = conn.command("POST", path, "glp", api_data=data)
        if resp["code"] != 202:
            conn.logger.error(f"Add device request failed for {inputs}!")
        else:
            conn.logger.info("Add device request accepted...")
        time.sleep(60 / POST_RPM)
        return resp

add_sub(conn, devices, sub, serial=False, key=False)

Add subscription to device(s).

API endpoint supports five devices per request. Handles chaining multiple requests for greater than five devices supplied. An additional response dict object will be appended to the return list for each additional request required to handle the number of input devices passed to the function.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
devices	list	List of device id(s) or serial numbers	required
sub	str	Subscription id or key	required
serial	bool	Flag to use device serial numbers, default is False	False
key	bool	Flag to use subscription key, default is False	False

Returns:

Type	Description
list[dict]	List of API response objects as provided by 'command' function in NewCentralBase

Source code in pycentral/glp/devices.py

def add_sub(self, conn, devices, sub, serial=False, key=False):
    """Add subscription to device(s).

    API endpoint supports five devices per request. Handles chaining multiple requests for
    greater than five devices supplied. An additional response dict object will be appended
    to the return list for each additional request required to handle the number of input
    devices passed to the function.

    Args:
        conn (NewCentralBase): pycentral base connection object
        devices (list): List of device id(s) or serial numbers
        sub (str): Subscription id or key
        serial (bool, optional): Flag to use device serial numbers, default is False
        key (bool, optional): Flag to use subscription key, default is False

    Returns:
        (list[dict]): List of API response objects as provided by 'command' function in NewCentralBase
    """

    if serial:
        d_list = []
        for d in devices:
            id = self.get_device_id(conn, d)
            if id[0]:
                d_list.append(id[1])
            else:
                conn.logger.error("Get device ID from serial failed!")
        devices = d_list

    if key:
        s = Subscriptions()
        id = s.get_sub_id(conn, sub)
        if id[0]:
            sub = id[1]
        else:
            conn.logger.error("Get sub ID from key failed!")

    split_input, wait_time = None, None

    # Split devices list per input size.
    if len(devices) > INPUT_SIZE:
        split_input, wait_time = rate_limit_check(
            devices, INPUT_SIZE, PATCH_RPM
        )
        conn.logger.info("WARNING MORE THAN 5 DEVICES IS A BETA FEATURE!")

    # Setup variables for iterating commands.
    queue = [devices] if not split_input else split_input
    resp_list = []
    path = generate_url(GLP_URLS["DEVICE"], category="devices")
    body = {"subscription": [{"id": sub}]}

    for inputs in queue:
        params = {"id": inputs}

        resp = conn.command(
            "PATCH", path, "glp", api_params=params, api_data=body
        )
        if resp["code"] == 202:
            conn.logger.info("Add sub request accepted...")
            id = resp["msg"]["transactionId"]
            status = check_progress(conn, id, self, limit=PATCH_RPM)
            if status[0]:
                conn.logger.info(
                    "Sucessfully added subscriptions to devices!"
                )
                resp_list.append(status[1])
            else:
                conn.logger.error("Add subscription failed!")
                resp_list.append(status[1])
        else:
            conn.logger.error("Bad request for add subscription!")
            resp_list.append(resp)

        if wait_time:
            time.sleep(wait_time)

    return resp_list

remove_sub(conn, devices, serial=False)

Remove a subscription from a device.

API endpoint supports five devices per request. Handles chaining multiple requests for greater than five devices supplied. An additional response dict object will be appended to the return list for each additional request required to handle the number of input devices passed to the function.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
devices	list	List of device id(s) or serial numbers	required
serial	bool	Flag to use device serial numbers, default is False	False

Returns:

Type	Description
list[dict]	List of API response objects as provided by 'command' function in NewCentralBase

Source code in pycentral/glp/devices.py

def remove_sub(self, conn, devices, serial=False):
    """Remove a subscription from a device.

    API endpoint supports five devices per request. Handles chaining multiple requests for
    greater than five devices supplied. An additional response dict object will be appended
    to the return list for each additional request required to handle the number of input
    devices passed to the function.

    Args:
        conn (NewCentralBase): pycentral base connection object
        devices (list): List of device id(s) or serial numbers
        serial (bool, optional): Flag to use device serial numbers, default is False

    Returns:
        (list[dict]): List of API response objects as provided by 'command' function in NewCentralBase
    """

    if serial:
        d_list = []
        for d in devices:
            id = self.get_device_id(conn, d)
            if id[0]:
                d_list.append(id[1])
            else:
                conn.logger.error("Get device ID from serial failed!")
        devices = d_list

    split_input, wait_time = None, None

    # Split devices list per input size.
    if len(devices) > INPUT_SIZE:
        split_input, wait_time = rate_limit_check(
            devices, INPUT_SIZE, PATCH_RPM
        )
        conn.logger.info("WARNING MORE THAN 5 DEVICES IS A BETA FEATURE!")

    # Setup variables for iterating commands.
    queue = [devices] if not split_input else split_input
    resp_list = []
    path = generate_url(GLP_URLS["DEVICE"], category="devices")
    body = {"subscription": []}

    for inputs in queue:
        params = {"id": inputs}

        resp = conn.command(
            "PATCH", path, "glp", api_params=params, api_data=body
        )
        if resp["code"] == 202:
            conn.logger.info("Remove sub request accepted...")
            id = resp["msg"]["transactionId"]
            status = check_progress(conn, id, self, limit=PATCH_RPM)
            if status[0]:
                conn.logger.info(
                    "Sucessfully Removed subscriptions from devices!"
                )
                resp_list.append(status[1])
            else:
                conn.logger.error("Remove subscription failed!")
                resp_list.append(status[1])
        else:
            conn.logger.error("Bad request for remove subscription!")
            resp_list.append(resp)

        if wait_time:
            time.sleep(wait_time)

    return resp_list

assign_devices(conn, devices=None, application=None, region=None, serial=False)

Assign devices to an application by passing one or more device id(s) or serial numbers.

Currently supports assigning and un-assigning devices to and from an application or applying/removing subscriptions to/from devices. Rate limits are enforced on this API. Five requests per minute is supported per workspace. API will result in 429 if this threshold is breached.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
devices	list	List of device id(s) or serial numbers	None
application	str	Application id	None
region	str	AHE region of the application the device is provisioned in	None
serial	bool	Flag to use device serial numbers, default is False	False

Returns:

Type	Description
dict	API response

Source code in pycentral/glp/devices.py

def assign_devices(
    self, conn, devices=None, application=None, region=None, serial=False
):
    """Assign devices to an application by passing one or more device id(s) or serial numbers.

    Currently supports assigning and un-assigning devices to and from an application or
    applying/removing subscriptions to/from devices. Rate limits are enforced on this API.
    Five requests per minute is supported per workspace. API will result in 429 if this
    threshold is breached.

    Args:
        conn (NewCentralBase): pycentral base connection object
        devices (list, optional):  List of device id(s) or serial numbers
        application (str, optional): Application id
        region (str, optional): AHE region of the application the device is provisioned in
        serial (bool, optional): Flag to use device serial numbers, default is False

    Returns:
        (dict): API response
    """

    conn.logger.info("Assigning device(s) to an application")
    path = generate_url(GLP_URLS["DEVICE"], category="devices")

    if serial:
        d_list = []
        for d in devices:
            id = self.get_device_id(conn, d)
            if id[0]:
                d_list.append(id[1])
            else:
                conn.logger.error("Get device ID from serial failed!")
        devices = d_list

    if len(devices) > INPUT_SIZE:
        resp = []
        rate_check = rate_limit_check(devices, INPUT_SIZE, PATCH_RPM)
        queue, wait_time = rate_check

        for i in range(len(queue)):
            params = {"id": queue[i]}

            data = {"application": {"id": application}, "region": region}

            time.sleep(wait_time)

            resp.append(
                conn.command(
                    api_method="PATCH",
                    api_path=path,
                    api_params=params,
                    api_data=data,
                    app_name="glp",
                )
            )

    else:
        params = {"id": devices}

        data = {"application": {"id": application}, "region": region}

        resp = conn.command(
            api_method="PATCH",
            api_path=path,
            api_params=params,
            api_data=data,
            app_name="glp",
        )

    if resp["code"] == 202:
        conn.logger.info(
            "Assign device(s) to application request accepted..."
        )
        id = resp["msg"]["transactionId"]
        status = check_progress(conn, id, self, limit=PATCH_RPM)
        if status[0]:
            conn.logger.info(
                "Sucessfully assigned device(s) to application!"
            )
            return status[1]
        else:
            conn.logger.error("Assign device(s) to application failed!")
            return status[1]
    conn.logger.error("Bad request for assign device(s) to application!")
    return resp

unassign_devices(conn, devices=None, serial=False)

Unassign devices from an application by passing one or more device id(s) or serial numbers.

Currently supports assigning and un-assigning devices to and from an application or applying/removing subscriptions to/from devices. Rate limits are enforced on this API. Five requests per minute is supported per workspace. API will result in 429 if this threshold is breached.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
devices	list	List of device id(s) or serial numbers	None
serial	bool	Flag to use device serial numbers, default is False	False

Returns:

Type	Description
dict	API response

Source code in pycentral/glp/devices.py

def unassign_devices(self, conn, devices=None, serial=False):
    """Unassign devices from an application by passing one or more device id(s) or serial numbers.

    Currently supports assigning and un-assigning devices to and from an application or
    applying/removing subscriptions to/from devices. Rate limits are enforced on this API.
    Five requests per minute is supported per workspace. API will result in 429 if this
    threshold is breached.

    Args:
        conn (NewCentralBase): pycentral base connection object
        devices (list, optional): List of device id(s) or serial numbers
        serial (bool, optional): Flag to use device serial numbers, default is False

    Returns:
        (dict): API response
    """

    conn.logger.info("Unassigning device(s) from an application")
    path = generate_url(GLP_URLS["DEVICE"], category="devices")

    if serial:
        d_list = []
        for d in devices:
            id = self.get_device_id(conn, d)
            if id[0]:
                d_list.append(id[1])
            else:
                conn.logger.error("Get device ID from serial failed!")
        devices = d_list

    if len(devices) > INPUT_SIZE:
        resp = []
        rate_check = rate_limit_check(devices, INPUT_SIZE, PATCH_RPM)
        queue, wait_time = rate_check

        for i in range(len(queue)):
            params = {"id": queue[i]}

            data = {"application": {"id": None}, "region": None}

            time.sleep(wait_time)

            resp.append(
                conn.command(
                    api_method="PATCH",
                    api_path=path,
                    api_params=params,
                    api_data=data,
                    app_name="glp",
                )
            )

    else:
        params = {"id": devices}

        data = {"application": {"id": None}, "region": None}

        resp = conn.command(
            api_method="PATCH",
            api_path=path,
            api_params=params,
            api_data=data,
            app_name="glp",
        )

    if resp["code"] == 202:
        conn.logger.info("Unassign device(s) from application accepted...")
        id = resp["msg"]["transactionId"]
        status = check_progress(conn, id, self, limit=PATCH_RPM)
        if status[0]:
            conn.logger.info(
                "Sucessfully unassigned device(s) from application!"
            )
            return status[1]
        else:
            conn.logger.error("Unassign device(s) from application failed!")
            return status[1]
    conn.logger.error(
        "Bad request for unassign device(s) from application!"
    )
    return resp

---

Service Manager

service_manager

ServiceManager

Bases: object

get_application_id_and_region(conn, application_name, region)

Retrieve the application ID and API region name for a specified application and region.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
application_name	str	name of the application to search for.	required
region	str	The region (UI name) where the application is deployed.	required

Returns:

Type	Description
dict	Dictionary containing the application ID and API region name if found, otherwise None.

Source code in pycentral/glp/service_manager.py

def get_application_id_and_region(self, conn, application_name, region):
    """Retrieve the application ID and API region name for a specified application and region.


    Args:
        conn (NewCentralBase): pycentral base connection object
        application_name (str): name of the application to search for.
        region (str): The region (UI name) where the application is deployed.

    Returns:
        (dict): Dictionary containing the application ID and API region name if found, otherwise None.
    """
    if not application_name or not region:
        conn.logger.error("Application name or region cannot be empty.")
        return None

    resp = self.get_service_manager_by_region(conn)
    if resp["code"] != 200:
        conn.logger.error(
            f"Error fetching list of service manager(applications) by region: {resp['code']} - {resp['msg']}"
        )
        return None

    region_service_manager_mapping = (
        self._generate_application_region_mapping(resp["msg"]["items"])
    )
    if region not in region_service_manager_mapping.keys():
        conn.logger.error(
            f"Unable to find region with name {region}. \nValid regions are {', '.join(region_service_manager_mapping.keys())}"
        )
        return None
    api_region_name = region_service_manager_mapping[region]["id"]

    region_service_managers = list(
        region_service_manager_mapping[region]["serviceManagers"].keys()
    )
    if application_name not in region_service_managers:
        conn.logger.error(
            f"Unable to find service manager with name {application_name}. \nValid service managers(applications) in region {region} are {', '.join(region_service_managers)}"
        )
        return None
    service_manager_id = region_service_manager_mapping[region][
        "serviceManagers"
    ][application_name]

    resp = self.get_service_manager_provisions(conn)
    if resp["code"] != 200:
        conn.logger.error(
            f"Error fetching list of service manager provisions (installed applications): {resp['code']} - {resp['msg']}"
        )
        return None
    for provisioned_service in resp["msg"]["items"]:
        if (
            service_manager_id
            == provisioned_service["serviceManager"]["id"]
            and provisioned_service["region"] == api_region_name
        ):
            conn.logger.info(
                f"Successfully verified installation of service manager {application_name} in region {region}."
            )
            return {
                "id": service_manager_id,
                "region": api_region_name,
            }

    conn.logger.error(
        f"Unable to find service manager(application) with name {application_name} in region {region}."
    )
    return None

get_service_manager_provisions(conn, limit=2000, offset=0)

Retrieve all provisioned services in GLP workspace.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required
limit	int	Specify the maximum number of entries per page. Maximum value accepted is 2000.	2000
offset	int	Specify pagination offset. Defines how many pages to skip before returning results. Default is 0.	0

Returns:

Type	Description
dict	API response containing provisioned services.

Source code in pycentral/glp/service_manager.py

def get_service_manager_provisions(self, conn, limit=2000, offset=0):
    """Retrieve all provisioned services in GLP workspace.

    Args:
        conn (NewCentralBase): pycentral base connection object
        limit (int, optional): Specify the maximum number of entries per page. Maximum value accepted is 2000.
        offset (int, optional): Specify pagination offset. Defines how many pages to skip before returning results. Default is 0.

    Returns:
        (dict): API response containing provisioned services.
    """
    conn.logger.info("Getting provisioned services in GLP workspace")
    path = generate_url(
        GLP_URLS["SERVICE_MANAGER_PROVISIONS"], category="service_catalog"
    )

    params = {
        "limit": limit,
        "offset": offset,
    }

    resp = conn.command(
        api_method="GET", api_path=path, api_params=params, app_name="glp"
    )
    return resp

get_service_manager_by_region(conn)

Get the region mapping for the service manager.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required

Returns:

Type	Description
dict	API response containing service managers by region.

Source code in pycentral/glp/service_manager.py

def get_service_manager_by_region(self, conn):
    """Get the region mapping for the service manager.

    Args:
        conn (NewCentralBase): pycentral base connection object

    Returns:
        (dict): API response containing service managers by region.
    """
    conn.logger.info("Getting services managers by region in GLP")
    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

get_service_managers(conn, limit=2000, offset=0)

Retrieve all available service managers in GLP.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object	required

limit (int, optional): Specify the maximum number of entries per page. Maximum value accepted is 2000. offset (int, optional): Specify pagination offset. Defines how many pages to skip before returning results.

Returns:

Type	Description
dict	API response containing service managers.

Source code in pycentral/glp/service_manager.py

def get_service_managers(self, conn, limit=2000, offset=0):
    """Retrieve all available service managers in GLP.

    Args:
        conn (NewCentralBase): pycentral base connection object
       limit (int, optional): Specify the maximum number of entries per page. Maximum value accepted is 2000.
       offset (int, optional): Specify pagination offset. Defines how many pages to skip before returning results.

    Returns:
        (dict): API response containing service managers.
    """
    conn.logger.info("Getting service managers in GLP")
    path = generate_url(
        GLP_URLS["SERVICE_MANAGER"], category="service_catalog"
    )

    params = {
        "limit": limit,
        "offset": offset,
    }

    resp = conn.command(
        api_method="GET", api_path=path, api_params=params, app_name="glp"
    )
    return resp

---

Subscriptions

subscriptions

Subscriptions

Bases: object

get_all_subscriptions(conn, select=None)

Get all subscriptions managed in a workspace.

API will result in 429 if this threshold is breached.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
select	list	A comma separated list of select properties to return in the response. By default, all properties are returned. Example: select=id,key	None

Returns:

Type	Description
list[dict]	A list of all subscriptions in the workspace, or an empty list if an error occurs.

Source code in pycentral/glp/subscriptions.py

def get_all_subscriptions(self, conn, select=None):
    """Get all subscriptions managed in a workspace.

    API will result in 429 if this threshold is breached.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        select (list, optional): A comma separated list of select properties to return in the response.
            By default, all properties are returned. Example: select=id,key

    Returns:
        (list[dict]): A list of all subscriptions in the workspace, or an empty list if an error occurs.
    """
    conn.logger.info("Getting all subscriptions in GLP workspace")

    limit = SUB_GET_LIMIT
    offset = 0
    count = 0
    result = []

    while True:
        resp = self.get_subscription(
            conn, limit=limit, offset=offset, select=select
        )
        if resp["code"] == 200:
            resp_message = resp["msg"]
            count += resp_message["count"]
            for subscription in resp_message["items"]:
                result.append(subscription)
            if count == resp_message["total"]:
                break
            else:
                offset += limit
        else:
            conn.logger.error(
                f"Error fetching list of subscriptions: {resp['code']} - {resp['msg']}"
            )
            result = []
            break

    return result

get_subscription(conn, filter=None, select=None, sort=None, limit=SUB_GET_LIMIT, offset=0)

Get a subscription managed in a workspace.

Rate limits are enforced on this API. 60 requests per minute is supported per workspace. API will result in 429 if this threshold is breached.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
filter	str	Filter expressions consisting of simple comparison operations joined by logical operators. Example: filter=key eq 'MHNBAP0001' and key in 'PAYHAH3YJE6THY, E91A7FDFE04D44C339'	None
select	list	A comma separated list of select properties to return in the response. By default, all properties are returned. Example: select=id,key	None
sort	str	A comma separated list of sort expressions. A sort expression is a property name optionally followed by a direction indicator asc or desc. Default is ascending order. Example: sort=key, quote desc	None
limit	int	Specifies the number of results to be returned. Default is 50. Range: 1-50.	SUB_GET_LIMIT
offset	int	Specifies the zero-based resource offset to start the response from. Default is 0.	0

Returns:

Type	Description
dict	API response containing subscriptions.

Source code in pycentral/glp/subscriptions.py

def get_subscription(
    self,
    conn,
    filter=None,
    select=None,
    sort=None,
    limit=SUB_GET_LIMIT,
    offset=0,
):
    """Get a subscription managed in a workspace.

    Rate limits are enforced on this API. 60 requests per minute is supported per workspace.
    API will result in 429 if this threshold is breached.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        filter (str, optional): Filter expressions consisting of simple comparison operations joined by logical operators.
            Example: filter=key eq 'MHNBAP0001' and key in 'PAYHAH3YJE6THY, E91A7FDFE04D44C339'
        select (list, optional): A comma separated list of select properties to return in the response.
            By default, all properties are returned. Example: select=id,key
        sort (str, optional): A comma separated list of sort expressions. A sort expression is a property name
            optionally followed by a direction indicator asc or desc. Default is ascending order.
            Example: sort=key, quote desc
        limit (int, optional): Specifies the number of results to be returned. Default is 50. Range: 1-50.
        offset (int, optional): Specifies the zero-based resource offset to start the response from. Default is 0.

    Returns:
        (dict): API response containing subscriptions.
    """
    conn.logger.info("Getting subscriptions in GLP workspace")
    path = generate_url(GLP_URLS["SUBSCRIPTION"], category="subscriptions")

    params = {"limit": limit, "offset": offset}
    if filter:
        params["filter"] = filter
    if select:
        params["select"] = select
    if sort:
        params["sort"] = sort

    resp = conn.command(
        api_method="GET", api_path=path, api_params=params, app_name="glp"
    )
    return resp

get_sub_id(conn, key)

Get subscription ID in a GLP workspace based on subscription key.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
key	str	Subscription key.	required

Returns:

Type	Description
tuple(bool, str)	Tuple of two elements. First element returns True if subscription ID is found, else False. The second element is a GLP subscription ID if found, else an error message.

Source code in pycentral/glp/subscriptions.py

def get_sub_id(self, conn, key):
    """Get subscription ID in a GLP workspace based on subscription key.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        key (str): Subscription key.

    Returns:
        (tuple(bool, str)): Tuple of two elements. First element returns True if subscription ID is found,
            else False. The second element is a GLP subscription ID if found, else an error message.
    """

    filter = f"key eq '{key}'"
    resp = self.get_subscription(conn, filter=filter)
    if resp["code"] != 200:
        return (resp, (False, "Bad request for get_sub"))
    elif resp["msg"]["count"] == 0:
        return (False, "Key not found")
    else:
        return (True, resp["msg"]["items"][0]["id"])

get_status(conn, id)

Get status of an async GLP subscription request.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
id	str	Transaction ID from async API request.	required

Returns:

Type	Description
dict	Response as provided by 'command' function in NewCentralBase.

Source code in pycentral/glp/subscriptions.py

def get_status(self, conn, id):
    """Get status of an async GLP subscription request.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        id (str): Transaction ID from async API request.

    Returns:
        (dict): Response as provided by 'command' function in NewCentralBase.
    """

    path = generate_url(
        f"{GLP_URLS['ASYNC']}/{id}", category="subscriptions"
    )
    resp = conn.command("GET", path, "glp")
    return resp

add_subscription(conn, subscriptions=None, limit=0, offset=0)

Add one or more subscriptions to a workspace.

This API provides an asynchronous response and will always return "202 Accepted" if basic input validations are successful. The location header in the response provides the URI to be invoked for fetching progress of the subscription addition task. For details about the status fetch URL, refer to the API Get progress or status of async operations in subscriptions. Rate limits are enforced on this API. 4 requests per minute is supported per workspace. API will result in 429 if this threshold is breached.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
subscriptions	list	List of subscription key objects. Example: [{"key": "string"}]	None
limit	int	Specifies the number of results to be returned. Default is 0. Range: 1-50.	0
offset	int	Specifies the zero-based resource offset to start the response from.	0

Returns:

Type	Description
dict	API response from the add subscription operation.

Source code in pycentral/glp/subscriptions.py

def add_subscription(self, conn, subscriptions=None, limit=0, offset=0):
    """Add one or more subscriptions to a workspace.

    This API provides an asynchronous response and will always return "202 Accepted" if basic input
    validations are successful. The location header in the response provides the URI to be invoked
    for fetching progress of the subscription addition task. For details about the status fetch URL,
    refer to the API Get progress or status of async operations in subscriptions.
    Rate limits are enforced on this API. 4 requests per minute is supported per workspace.
    API will result in 429 if this threshold is breached.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        subscriptions (list, optional): List of subscription key objects.
            Example: [{"key": "string"}]
        limit (int, optional): Specifies the number of results to be returned. Default is 0. Range: 1-50.
        offset (int, optional): Specifies the zero-based resource offset to start the response from.

    Returns:
        (dict): API response from the add subscription operation.
    """
    conn.logger.info("Adding subscription(s) to GLP workspace")
    path = generate_url(GLP_URLS["SUBSCRIPTION"], category="subscriptions")

    if len(subscriptions) > INPUT_SIZE:
        resp = []
        rate_check = rate_limit_check(subscriptions, INPUT_SIZE, POST_RPM)
        queue, wait_time = rate_check

        for i in range(len(queue)):
            params = {
                "offset": offset,
            }

            data = {"subscriptions": queue[i]}

            time.sleep(wait_time)

            resp.append(
                conn.command(
                    api_method="POST",
                    api_path=path,
                    api_params=params,
                    api_data=data,
                    app_name="glp",
                )
            )

    else:
        params = {
            "offset": offset,
        }

        data = {"subscriptions": subscriptions}

        resp = conn.command(
            api_method="POST",
            api_path=path,
            api_params=params,
            api_data=data,
            app_name="glp",
        )

    if resp["code"] == 202:
        conn.logger.info(
            "Add subscription(s) to workspace request accepted..."
        )
        id = resp["msg"]["transactionId"]
        status = check_progress(conn, id, self, limit=SUB_LIMIT)
        if status[0]:
            conn.logger.info(
                "Sucessfully added subscription(s) to workspace!"
            )
            return status[1]
        else:
            conn.logger.error("Add subscription(s) to workspace failed!")
            return status[1]
    conn.logger.error("Bad request for add subscription(s) to workspace!")
    return resp

---

User Management

user_management

UserMgmt

Bases: object

get_users(conn, filter=None, limit=300, offset=0)

Retrieve users that match given filters.

All users are returned when no filters are provided. The Get users API can be filtered by: id, username, userStatus, createdAt, updatedAt, lastLogin.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
filter	str	Filter data using a subset of OData 4.0 and return only the subset of resources that match the filter. Examples: Filter with id: filter=id eq '7600415a-8876-5722-9f3c-b0fd11112283' Filter with username: filter=username eq 'user@example.com' Filter with userStatus: filter=userStatus neq 'UNVERIFIED' Filter with createdAt: filter=createdAt gt '2020-09-21T14:19:09.769747' Filter with updatedAt: filter=updatedAt gt '2020-09-21T14:19:09.769747' Filter with lastLogin: filter=lastLogin lt '2020-09-21T14:19:09.769747'	None
limit	int	Specify the maximum number of entries per page. Maximum value accepted is 300. Range: 1-300.	300
offset	int	Specify pagination offset. Defines how many pages to skip before returning results.	0

Returns:

Type	Description
dict	API response containing users.

Source code in pycentral/glp/user_management.py

def get_users(self, conn, filter=None, limit=300, offset=0):
    """Retrieve users that match given filters.

    All users are returned when no filters are provided. The Get users API can be filtered by:
    id, username, userStatus, createdAt, updatedAt, lastLogin.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        filter (str, optional): Filter data using a subset of OData 4.0 and return only the subset of
            resources that match the filter. Examples:

            - Filter with id: filter=id eq '7600415a-8876-5722-9f3c-b0fd11112283'
            - Filter with username: filter=username eq 'user@example.com'
            - Filter with userStatus: filter=userStatus neq 'UNVERIFIED'
            - Filter with createdAt: filter=createdAt gt '2020-09-21T14:19:09.769747'
            - Filter with updatedAt: filter=updatedAt gt '2020-09-21T14:19:09.769747'
            - Filter with lastLogin: filter=lastLogin lt '2020-09-21T14:19:09.769747'
        limit (int, optional): Specify the maximum number of entries per page. Maximum value accepted is 300. Range: 1-300.
        offset (int, optional): Specify pagination offset. Defines how many pages to skip before returning results. 

    Returns:
        (dict): API response containing users.
    """
    conn.logger.info("Getting users in GLP workspace")
    path = generate_url(
        GLP_URLS["USER_MANAGEMENT"], category="user_management"
    )

    params = {
        "limit": limit,
        "offset": offset,
    }
    if filter:
        params["filter"] = filter

    resp = conn.command(
        api_method="GET", api_path=path, api_params=params, app_name="glp"
    )
    return resp

get_user(conn, email=None, id=None)

Get a user from a workspace.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
email	str	Account username (email address).	None
id	str	Target user ID.	None

Returns:

Type	Description
dict	Response as provided by 'command' function in NewCentralBase.

Source code in pycentral/glp/user_management.py

def get_user(self, conn, email=None, id=None):
    """Get a user from a workspace.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        email (str, optional): Account username (email address).
        id (str, optional): Target user ID.

    Returns:
        (dict): Response as provided by 'command' function in NewCentralBase.
    """
    conn.logger.info("Getting a user in GLP workspace")
    if email:
        id = self.get_user_id(conn, email)[1]

    path = generate_url(
        f"{GLP_URLS['USER_MANAGEMENT']}/{id}", category="user_management"
    )

    resp = conn.command("GET", path, "glp")
    if resp["code"] == 200:
        conn.logger.info("Get user successful!")
    else:
        conn.logger.error("Get user failed!")
    return resp

get_user_id(conn, email)

Get user ID in a GLP workspace by email.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
email	str	Account username (email address).	required

Returns:

Type	Description
tuple(bool, str)	Tuple of two elements. First element returns True if user ID is found, else False. The second element is a GLP user ID if found, else an error message.

Source code in pycentral/glp/user_management.py

def get_user_id(self, conn, email):
    """Get user ID in a GLP workspace by email.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        email (str): Account username (email address).

    Returns:
        (tuple(bool, str)): Tuple of two elements. First element returns True if user ID is found,
            else False. The second element is a GLP user ID if found, else an error message.
    """

    filter = f"username eq '{email}'"
    resp = self.get_users(conn, filter=filter)
    if resp["code"] != 200:
        return (resp, (False, "Bad request for get_id"))
    elif resp["msg"]["count"] == 0:
        return (False, "Email not found")
    else:
        return (True, resp["msg"]["items"][0]["id"])

delete_user(conn, email=None, user_id=None)

Delete a user from a workspace.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
email	str	Account username (email address).	None
user_id	str	Target user ID.	None

Returns:

Type	Description
dict	API response from the delete operation.

Source code in pycentral/glp/user_management.py

def delete_user(self, conn, email=None, user_id=None):
    """Delete a user from a workspace.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        email (str, optional): Account username (email address).
        user_id (str, optional): Target user ID.

    Returns:
        (dict): API response from the delete operation.
    """
    conn.logger.info("Deleting a user in GLP workspace")
    if email:
        user_id = self.get_user_id(conn, email)[1]

    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

inv_user(conn, email, send_link)

Invite a user to a GLP workspace.

Parameters:

Name	Type	Description	Default
conn	NewCentralBase	pycentral base connection object.	required
email	str	Email address of the user to invite.	required
send_link	bool	Set to True to send welcome email.	required

Returns:

Type	Description
dict	Response as provided by 'command' function in NewCentralBase.

Source code in pycentral/glp/user_management.py

def inv_user(self, conn, email, send_link):
    """Invite a user to a GLP workspace.

    Args:
        conn (NewCentralBase): pycentral base connection object.
        email (str): Email address of the user to invite.
        send_link (bool): Set to True to send welcome email.

    Returns:
        (dict): Response as provided by 'command' function in NewCentralBase.
    """
    conn.logger.info("Inviting a user to GLP workspace")

    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"] == 201:
        conn.logger.info("Invite user successful!")
    else:
        conn.logger.error("Invite user failed!")
    return resp