Base Module

The base module provides the core connection functionality for PyCentral.

NewCentralBase

NewCentralBase(token_info, logger=None, log_level='INFO', enable_scope=False)

Constructor initializes the NewCentralBase class with token information and logging configuration.

Validates and processes the provided token information, sets up logging, and optionally initializes scope-related functionality.

Parameters:

Name	Type	Description	Default
token_info	dict or str	Dictionary containing token information for supported applications (new_central, glp). Can also be a string path to a YAML or JSON file with token information.	required
logger	Logger	Logger instance. Defaults to None.	None
log_level	str	Logging level. Defaults to "INFO".	'INFO'
enable_scope	bool	Flag to enable scope management. If True, the SDK will automatically fetch data about existing scopes and associated profiles, simplifying scope and configuration management. If False, scope-related API calls are disabled, resulting in faster initialization. Defaults to False.	False

Source code in pycentral/base.py

def __init__(self, token_info, logger=None, log_level="INFO", enable_scope=False):
    """
    Constructor initializes the NewCentralBase class with token information and logging configuration.

    Validates and processes the provided token information, sets up logging,
    and optionally initializes scope-related functionality.

    Args:
        token_info (dict or str): Dictionary containing token information for supported
            applications (new_central, glp). Can also be a string path to a YAML or
            JSON file with token information.
        logger (logging.Logger, optional): Logger instance. Defaults to None.
        log_level (str, optional): Logging level. Defaults to "INFO".
        enable_scope (bool, optional): Flag to enable scope management. If True, the SDK
            will automatically fetch data about existing scopes and associated profiles,
            simplifying scope and configuration management. If False, scope-related API
            calls are disabled, resulting in faster initialization. Defaults to False.
    """
    self.token_info = new_parse_input_args(token_info)
    self.token_file_path = None
    if isinstance(token_info, str):
        self.token_file_path = token_info
    self.logger = self.set_logger(log_level, logger)
    self._app_routes = self._build_app_routes()
    self.scopes = None
    self._http_clients = {}
    self._initialize_http_clients()
    self._initialize_tokens()
    if enable_scope:
        self.scopes = Scopes(central_conn=self)

set_logger(log_level, logger=None)

Set up the logger.

Parameters:

Name	Type	Description	Default
log_level	str	Logging level.	required
logger	Logger	Logger instance. Defaults to None.	None

Returns:

Type	Description
Logger	Logger instance.

Source code in pycentral/base.py

def set_logger(self, log_level, logger=None):
    """
    Set up the logger.

    Args:
        log_level (str): Logging level.
        logger (logging.Logger, optional): Logger instance. Defaults to None.

    Returns:
        (logging.Logger): Logger instance.
    """
    if logger:
        return logger
    else:
        return console_logger("NEW CENTRAL BASE", log_level)

create_token(app_name)

Create a new access token for the specified application.

Generates a new access token using the client credentials for the specified application, updates the self.token_info dictionary with the new token, and optionally saves it to a file.

Parameters:

Name	Type	Description	Default
app_name	str	Name of the application. Supported applications: "new_central", "glp".	required

Returns:

Type	Description
str	Access token.

Raises:

Type	Description
LoginError	If there is an error during token creation.

Source code in pycentral/base.py

def create_token(self, app_name):
    """
    Create a new access token for the specified application.

    Generates a new access token using the client credentials for the specified
    application, updates the self.token_info dictionary with the new token,
    and optionally saves it to a file.

    Args:
        app_name (str): Name of the application. Supported applications: "new_central", "glp".

    Returns:
        (str): Access token.

    Raises:
        LoginError: If there is an error during token creation.
    """
    client_id, client_secret = self._return_client_credentials(app_name)
    client = BackendApplicationClient(client_id)

    oauth = OAuth2Session(client=client)
    auth = HTTPBasicAuth(client_id, client_secret)

    token_url = self.token_info[app_name].get("_token_url")
    if not token_url:
        raise ValueError(
            f"Cannot determine token URL for '{app_name}'. "
            "Ensure valid credentials (including workspace_id for unified mode) are provided."
        )

    try:
        self.logger.info(f"Attempting to create new token from {app_name}")
        token = oauth.fetch_token(token_url=token_url, auth=auth)
        if "access_token" not in token:
            msg = (
                f"Token response for '{app_name}' did not contain an access_token. "
                "Verify that the client credentials and token URL are correct."
            )
            self.logger.error(msg)
            raise LoginError(msg)
        self.logger.info(
            f"{app_name} Login Successful.. Obtained Access Token!"
        )
        self.token_info[app_name]["access_token"] = token["access_token"]
        if self.token_file_path:
            save_access_token(
                app_name,
                token["access_token"],
                self.token_file_path,
                self.logger,
            )
        return token["access_token"]
    except Exception as e:
        # unified extraction of status code (from exception or its response)
        status_code = getattr(e, "status_code", None)
        resp = getattr(e, "response", None)
        if resp is not None:
            status_code = getattr(resp, "status_code", status_code)

        # special-case invalid client credentials to provide a clearer, actionable message
        if isinstance(e, InvalidClientError):
            description = getattr(e, "description", None) or str(e)
            msg = (
                f"{description} for {app_name}. "
                "Provide valid client_id and client_secret to create an access token."
            )
        else:
            msg = str(e) or "Unexpected error while creating access token"

        self.logger.error(msg)
        raise LoginError(msg, status_code)

command(api_method, api_path, app_name='new_central', api_data=None, api_params=None, headers=None, files=None)

Execute an API command to Central or GreenLake Platform.

This is the primary method for making API calls from the SDK. It handles authentication, token refresh on expiry, request formatting, and response parsing. All other SDK modules internally use this method to make API calls.

The method automatically

• Validates the application name and HTTP method
• Constructs the full URL from self.base_url and api_path
• Adds appropriate headers (Content-Type, Accept) if not provided
• Serializes api_data to JSON when Content-Type is application/json
• Handles 401 errors by refreshing the access token and retrying if client credentials are available
• Parses JSON responses when possible

Parameters:

Name	Type	Description	Default
api_method	str	HTTP method for the API call. Supported methods: POST, PATCH, DELETE, GET, PUT.	required
api_path	str	API endpoint path (e.g., "monitoring/v1/aps"). This is appended to the base_url configured in token_info.	required
app_name	str	Target application for the API call. Use "new_central" for Central APIs (default). Use "glp" for GreenLake Platform APIs.	'new_central'
api_data	dict	Request body/payload to be sent. Automatically serialized to JSON if Content-Type is application/json. Defaults to None.	None
api_params	dict	URL query parameters for the API request. Defaults to None.	None
headers	dict	Custom HTTP headers. If not provided and no files are being uploaded, defaults to {"Content-Type": "application/json", "Accept": "application/json"}.	None
files	dict	Files to upload in multipart/form-data requests. When provided, Content-Type header is not automatically set. Defaults to None.	None

Returns:

Type	Description
dict	API response containing: - code (int): HTTP status code - msg (dict or str): Parsed JSON response body, or raw text if not JSON - headers (dict): Response headers

Raises:

Type	Description
ResponseError	If there is an error during the API call.

Source code in pycentral/base.py

def command(
    self,
    api_method,
    api_path,
    app_name="new_central",
    api_data=None,
    api_params=None,
    headers=None,
    files=None,
):
    """
    Execute an API command to Central or GreenLake Platform.

    This is the primary method for making API calls from the SDK. It handles
    authentication, token refresh on expiry, request formatting, and response
    parsing. All other SDK modules internally use this method to make API calls.

    The method automatically:
        - Validates the application name and HTTP method
        - Constructs the full URL from self.base_url and api_path
        - Adds appropriate headers (Content-Type, Accept) if not provided
        - Serializes api_data to JSON when Content-Type is application/json
        - Handles 401 errors by refreshing the access token and retrying if client credentials are available
        - Parses JSON responses when possible

    Args:
        api_method (str): HTTP method for the API call. Supported methods:
            POST, PATCH, DELETE, GET, PUT.
        api_path (str): API endpoint path (e.g., "monitoring/v1/aps").
            This is appended to the base_url configured in token_info.
        app_name (str, optional): Target application for the API call.
            Use "new_central" for Central APIs (default).
            Use "glp" for GreenLake Platform APIs.
        api_data (dict, optional): Request body/payload to be sent. Automatically
            serialized to JSON if Content-Type is application/json. Defaults to None.
        api_params (dict, optional): URL query parameters for the API request.
            Defaults to None.
        headers (dict, optional): Custom HTTP headers. If not provided and no files
            are being uploaded, defaults to {"Content-Type": "application/json",
            "Accept": "application/json"}.
        files (dict, optional): Files to upload in multipart/form-data requests.
            When provided, Content-Type header is not automatically set. Defaults to None.

    Returns:
        (dict): API response containing:
            - code (int): HTTP status code
            - msg (dict or str): Parsed JSON response body, or raw text if not JSON
            - headers (dict): Response headers

    Raises:
        ResponseError: If there is an error during the API call.
    """
    self._validate_request(app_name, api_method)

    api_data = api_data if api_data is not None else {}
    files = files if files is not None else {}

    api_params = self._prepare_query_params(api_params)
    req_headers = self._build_request_headers(headers, files)
    req_data = self._serialize_request_data(api_data, req_headers)

    route = self._app_routes[app_name]
    url = build_url(route["base_url"], api_path)

    retry = 0
    resp = None
    limit_reached = False
    try:
        while not limit_reached:
            resp = self.request_url(
                url=url,
                data=req_data,
                method=api_method,
                headers=req_headers,
                params=api_params,
                files=files,
                access_token=self.token_info[route["token_key"]]["access_token"],
                app_name=app_name,
            )
            if resp.status_code == 401:
                if retry >= 1:
                    self.logger.error(
                        "Received error 401 on requesting url "
                        "%s with resp %s" % (str(url), str(resp.text))
                    )
                    limit_reached = True
                    break
                self.logger.info(
                    f"{app_name} access token has expired. Handling Token Expiry..."
                )
                self._renew_token(route["token_key"])
                retry += 1
            elif resp.status_code == 429:
                # Allowing one retry on 429 in case rate limit hit
                # Pausing 1 second should allow the retry to succeed
                # if it doesn't, we log the error and exit
                if retry >= 2:
                    self.logger.error(
                        f"Received error 429 on requesting url "
                        f"{url} with resp {resp.text}. Retry attempts exhausted."
                    )
                    limit_reached = True
                    break
                self.logger.info(
                    f"Received error 429 - {app_name} rate limit reached."
                    f" Pausing before retry {retry + 1}/2..."
                )
                time.sleep(1)
                retry += 1
            else:
                break

        if resp is None:
            raise ResponseError(
                f"{api_method} FAILURE ",
                RuntimeError(f"No response returned for {api_method} {url}"),
            )

        result = {
            "code": resp.status_code,
            "msg": resp.text,
            "headers": dict(resp.headers),
        }

        try:
            result["msg"] = resp.json()
        except (ValueError, json.JSONDecodeError):
            result["msg"] = resp.text

        return result

    except (LoginError, ValueError):
        raise
    except Exception as err:
        err_str = f"{api_method} FAILURE "
        self.logger.error(err)
        raise ResponseError(err_str, err)

request_url(url, access_token, app_name='new_central', data=None, method='GET', headers=None, params=None, files=None)

Make an API call to application (New Central or GLP) via the requests library.

Parameters:

Name	Type	Description	Default
url	str	HTTP Request URL string.	required
access_token	str	Access token for authentication.	required
app_name	str	App name used to select HTTP client. Defaults to "new_central".	'new_central'
data	dict	HTTP Request payload. Defaults to None.	None
method	str	HTTP Request Method supported by GLP/New Central. Defaults to "GET".	'GET'
headers	dict	HTTP Request headers. Defaults to None.	None
params	dict	HTTP URL query parameters. Defaults to None.	None
files	dict	Files dictionary with file pointer depending on API endpoint as accepted by GLP/New Central. Defaults to None.	None

Returns:

Type	Description
Response	HTTP response of API call using requests library.

Raises:

Type	Description
ResponseError	If there is an error during the API call.

Note

Transient transport failures (DNS resolution errors, connection failures, timeouts, and proxy errors) are automatically retried with exponential backoff up to RETRY_MAX_RETRIES attempts. All other exceptions are raised immediately as ResponseError without retrying.

Source code in pycentral/base.py

def request_url(
    self,
    url,
    access_token,
    app_name="new_central",
    data=None,
    method="GET",
    headers=None,
    params=None,
    files=None,
):
    """Make an API call to application (New Central or GLP) via the requests library.

    Args:
        url (str): HTTP Request URL string.
        access_token (str): Access token for authentication.
        app_name (str, optional): App name used to select HTTP client.
            Defaults to "new_central".
        data (dict, optional): HTTP Request payload. Defaults to None.
        method (str, optional): HTTP Request Method supported by GLP/New Central.
            Defaults to "GET".
        headers (dict, optional): HTTP Request headers. Defaults to None.
        params (dict, optional): HTTP URL query parameters. Defaults to None.
        files (dict, optional): Files dictionary with file pointer depending on
            API endpoint as accepted by GLP/New Central. Defaults to None.

    Returns:
        (requests.models.Response): HTTP response of API call using requests library.

    Raises:
        ResponseError: If there is an error during the API call.

    Note:
        Transient transport failures (DNS resolution errors, connection failures,
        timeouts, and proxy errors) are automatically retried with exponential
        backoff up to ``RETRY_MAX_RETRIES`` attempts. All other exceptions are
        raised immediately as ``ResponseError`` without retrying.
    """
    req_headers = dict(headers) if headers else {}
    req_headers["authorization"] = "Bearer " + access_token

    # Build keyword args — httpx does not allow content + data + files together
    kwargs = {
        "method": method,
        "url": url,
        "headers": req_headers,
    }
    if params:
        kwargs["params"] = params
    if files:
        # Multipart upload: use files + optional form data (dict only)
        kwargs["files"] = files
        if data and isinstance(data, dict):
            kwargs["data"] = data
    elif isinstance(data, str):
        # Pre-serialized JSON string — pass directly, httpx handles encoding
        kwargs["content"] = data
    elif isinstance(data, bytes):
        # Raw bytes
        kwargs["content"] = data
    elif isinstance(data, dict) and data:
        # Form-encoded dict
        kwargs["data"] = data

    http_client = self._http_clients.get(app_name)
    if http_client is None:
        http_client = self._create_http_client(app_name)
        self._http_clients[app_name] = http_client

    retry_count = 0
    while True:
        try:
            return http_client.request(**kwargs)
        except TRANSIENT_TRANSPORT_ERRORS as err:
            if retry_count >= RETRY_MAX_RETRIES:
                err_str = f"Failed making request to URL {url} with error {err}"
                self.logger.error(err_str)
                raise ResponseError(err_str, err)
            retry_count += 1
            delay = min(
                RETRY_INITIAL_BACKOFF * (RETRY_BACKOFF_MULTIPLIER ** (retry_count - 1)),
                RETRY_MAX_BACKOFF,
            )
            self.logger.warning(
                "Transient transport failure on %s %s for app %s "
                "(retry %s/%s): %s. Retrying in %.1f seconds.",
                method, url, app_name, retry_count, RETRY_MAX_RETRIES, err, delay,
            )
            time.sleep(delay)
        except Exception as err:
            err_str = f"Failed making request to URL {url} with error {err}"
            self.logger.error(err_str)
            raise ResponseError(err_str, err)

get_scopes()

Set up the scopes for the current instance by creating a Scopes object.

Initializes the scopes attribute using the Scopes class, passing the current instance as the central_conn parameter. If the scopes attribute is already initialized, it simply returns the existing object.

Returns:

Type	Description
Scopes	The initialized or existing Scopes object.

Source code in pycentral/base.py

def get_scopes(self):
    """
    Set up the scopes for the current instance by creating a Scopes object.

    Initializes the scopes attribute using the Scopes class, passing the
    current instance as the central_conn parameter. If the scopes attribute
    is already initialized, it simply returns the existing object.

    Returns:
        (Scopes): The initialized or existing Scopes object.
    """
    if self.scopes is None:
        self.scopes = Scopes(central_conn=self)
    return self.scopes

close()

Close all underlying HTTP clients and release connection pool resources.

Source code in pycentral/base.py

def close(self):
    """Close all underlying HTTP clients and release connection pool resources."""
    for app_name, http_client in self._http_clients.items():
        try:
            if http_client:
                http_client.close()
        except Exception as err:
            self.logger.error(f"Failed closing HTTP client for {app_name}: {err}")

    self._http_clients = {}