### GET request setup with exception handling Source: https://pyedgeconnect.readthedocs.io/en/latest/_modules/pyedgeconnect.html Sets up an HTTP GET request and sends the response to _handle_response. The docstring documents parameters and return behavior, but the method body is not shown in the source. Catches exceptions and logs errors similarly to _post. ```python def _get( self, api_path: str, expected_status: list = [200], return_type: str = "json", ): """Setup HTTP GET request and send results to _handle_response method. Catches Exceptions and logs to log file :param api_path: API path to append to url_prefix :type api_path: str :param expected_status: List of expected HTTP status codes of response, defaults to [200] :type expected_status: list, optional :param return_type: Filter for data to include in response to function call, accepted values are "json" "text" "bool" "full_response", defaults to "json" :type return_type: str, optional :return: Returns False on exceptions, otherwise passes return through _handle_response method for processing Requests response :rtype: bool, _handle_response method """ ``` -------------------------------- ### Clone pyedgeconnect repository to download examples Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/index.html Use this git clone command to download the pyedgeconnect repository, which contains all documented example scripts in the examples folder. ```bash $ git clone https://github.com/aruba/pyedgeconnect.git ``` -------------------------------- ### Start demo with docker-compose up -d Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/ec_telemetry_demo.html Starts the demo infrastructure and begins telemetry collection. Requires a configured .env file or host environment variables pointing to an Orchestrator and EdgeConnect appliances. If the user is not in the docker group and uses sudo, add the -E flag to pass host environment variables. ```bash $ docker-compose up -d ``` -------------------------------- ### Start demo with monitoring compose files Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/ec_telemetry_demo.html Starts the demo along with optional monitoring containers (cAdvisor and Prometheus) by specifying multiple compose files. A Grafana dashboard is included to visualize container resource utilization. ```bash $ docker-compose -f docker-compose.yml -f docker-compose.monitor.yml up -d ``` -------------------------------- ### Install requirements for boost_timeseries example Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/boost_timeseries.html Installs the Python packages listed in requirements.txt, including pandas, which is required by the boost_timeseries script. ```bash pip install -r requirements.txt ``` -------------------------------- ### Install required dependencies Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/generate_preconfig.html Install the necessary Python packages, including pyedgeconnect and jinja2, to run the preconfig generation scripts. ```bash pip install -r requirements.txt # OR pip install pyedgeconnect pip install jinja2 ``` -------------------------------- ### Install a specific branch from GitHub Source: https://pyedgeconnect.readthedocs.io/en/latest/install.html Shows how to install a specific branch of pyedgeconnect from GitHub using the @branch syntax, with an example for the development branch. ```bash $ pip install git+https://github.com/aruba/pyedgeconnect@ # Install the Development branch $ pip install git+https://github.com/aruba/pyedgeconnect@development ``` -------------------------------- ### Build documentation locally Source: https://pyedgeconnect.readthedocs.io/en/latest/install.html Clones the pyedgeconnect repository, installs with the [dev] option to include Sphinx and related packages, and builds the documentation with make html. ```bash $ git clone https://github.com/aruba/pyedgeconnect.git $ cd pyedgeconnect $ pip install .[dev] $ cd docs $ make html ``` -------------------------------- ### Rebuild and start demo with docker-compose up -d --build Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/ec_telemetry_demo.html Rebuilds containers after changes to scripts or files in the app/ec-telemetry directory (e.g., limit_appliances.json) and starts the demo. Use this when you modify those files. ```bash $ docker-compose up -d --build ``` -------------------------------- ### Type hinting examples (PEP585) Source: https://pyedgeconnect.readthedocs.io/en/latest/code_formatting.html Examples of function definitions with type hints using PEP585 generic types (e.g., `list[int]`). ```python def my_func(my_var: str) -> dict: ... # or def my_func(my_var: list[int]) -> list: ... ``` -------------------------------- ### POST request setup with exception handling Source: https://pyedgeconnect.readthedocs.io/en/latest/_modules/pyedgeconnect.html Sets up an HTTP POST request and sends the response to _handle_response. Validates the return_type against accepted values and logs an error for unknown types. Catches ConnectTimeout and general exceptions, logging errors and returning False on failure. ```python def _post( self, api_path: str, data="", files={}, expected_status: list = [200], return_type: str = "json", ): """Setup HTTP POST request and send results to _handle_response method. Catches Exceptions and logs to log file :param api_path: API path to append to url_prefix :type api_path: str :param data: Data to pass in request body, defaults to "" :type data: str, list, dict, optional :param expected_status: List of expected HTTP status codes of response, defaults to [200] :type expected_status: list, optional :param return_type: Filter for data to include in response to function call, accepted values are "json" "text" "bool" "full_response", defaults to "json" :type return_type: str, optional :return: Returns False on exceptions, otherwise passes return through _handle_response method for processing Requests response :rtype: bool, _handle_response method """ if return_type not in ["json", "text", "bool", "full_response"]: self.logger.error( "Called POST {} with unknown return type '{}'".format( api_path, return_type ) ) try: response = self._req_post(api_path, data, files) return self._handle_response( api_path, response, expected_status, return_type ) except requests.exceptions.ConnectTimeout: self.logger.error( f"POST {api_path} | Request Timed Out - " f"Timeout values (connect/read): {self.timeout}" ) except Exception as ex: self.logger.error( "Exception {} when calling POST {}. Traceback: {}".format( type(ex), api_path, traceback.format_exc() ) ) return False ``` -------------------------------- ### Install pyedgeconnect from GitHub Source: https://pyedgeconnect.readthedocs.io/en/latest/install.html Installs the most recent version of pyedgeconnect directly from the GitHub repository using pip. The output shows the package version in the pip list. ```bash $ pip install git+https://github.com/aruba/pyedgeconnect ... $ pip list Package Version ----------------------------- -------------------------------- ... ... pyedgeconnect x.y.z ... ... ``` -------------------------------- ### Install pyedgeconnect from PyPI Source: https://pyedgeconnect.readthedocs.io/en/latest/install.html Installs the latest pyedgeconnect package from PyPI using pip. The output shows the package version in the pip list. ```bash $ pip install pyedgeconnect ... $ pip list Package Version ----------------------------- -------------------------------- ... ... pyedgeconnect x.y.z ... ... ``` -------------------------------- ### Authenticate to Orchestrator Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/auth_example.html Example script demonstrating how to connect to an Orchestrator instance using API keys or username/password credentials, with support for environment variables and MFA. ```python import argparse import getpass import os from pyedgeconnect import Orchestrator # Parse runtime arguments to specify an Orchestrator IP or FQDN parser = argparse.ArgumentParser() parser.add_argument( "-o", "--orch", help="specify Orchestrator URL", type=str, ) args = parser.parse_args() # Set Orchestrator FQDN/IP via arguments, environment variable, # or user input if neither in argument or environment variable if vars(args)["orch"] is not None: orch_url = vars(args)["orch"] elif os.getenv("ORCH_URL") is not None: orch_url = os.getenv("ORCH_URL") else: orch_url = input("Orchstrator IP or FQDN: ") # Set Orchestrator API Key via environment variable or user input # Skipping will fallback to user/password authentication if os.getenv("ORCH_API_KEY") is not None: orch_api_key = os.getenv("ORCH_API_KEY") else: orch_api_key_input = input("Orchstrator API Key (enter to skip): ") if len(orch_api_key_input) == 0: orch_api_key = None # Set user and password if present in environment variable orch_user = os.getenv("ORCH_USER") orch_pw = os.getenv("ORCH_PASSWORD") else: orch_api_key = orch_api_key_input # Instantiate Orchestrator with ``log_console`` enabled for # printing log messages to terminal, and ``verify_ssl`` to False # which will not verify the web https certificate on Orchestrator orch = Orchestrator( orch_url, api_key=orch_api_key, log_console=True, verify_ssl=False, ) # If not using API key, login to Orchestrator with username/password if orch_api_key is None: # If username/password not in environment variables, prompt user if orch_user is None: orch_user = input("Enter Orchestrator username: ") orch_pw = getpass.getpass("Enter Orchestrator password: ") # Check if multi-factor authentication required mfa_prompt = input("Are you using MFA for this user (y/n)?: ") if mfa_prompt == "y": orch.send_mfa(orch_user, orch_pw, temp_code=False) token = input("Enter MFA token: ") else: token = "" # Login to Orchestrator with user/password to check auth before # proceeding confirm_auth = orch.login(orch_user, orch_pw, mfacode=token) if confirm_auth: pass else: print("Authentication to Orchestrator Failed") exit() # If API key specified, check that key is valid before proceeding else: confirm_auth = orch.get_orchestrator_hello() if confirm_auth != "There was an internal server error.": pass else: print("Authentication to Orchestrator Failed") exit() ``` -------------------------------- ### Configure new service and associate all appliances Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/ec_sse_orchestration.html Runs the script to configure a new service, associate all appliances, and configure locations/tunnels on the SSE service. ```bash python ec_sse_quickstart.py -af all -cso -as -sse ``` -------------------------------- ### Create and activate a Python virtual environment Source: https://pyedgeconnect.readthedocs.io/en/latest/install.html Sets up a Python 3.7+ virtual environment using python3.9 and activates it. The output shows the Python version. This is recommended to avoid conflicts with other Python versions. ```bash :~$ python3.9 -m venv my_new_project :~$ source my_new_project/bin/activate (my_new_project) :~$ python3 --version Python 3.9.13 ``` -------------------------------- ### Login and Logout with Credentials Source: https://pyedgeconnect.readthedocs.io/en/latest/quickstart.html Connect to the instance using the login method with username and password, then log out when done. ```python orch.login("username", "password") orch.logout() # or ec.login("username", "password") ec.logout() ``` -------------------------------- ### Run wan_util_95th.py with runtime arguments Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/wan_util_95th.html Run the wan_util_95th.py script with different runtime arguments. Use -c to collect data for all appliances, -c -a to collect for a single appliance, -r to report on existing data files, and -c -d to update deployment interface comments. The -c or -r argument is required. ```bash python wan_util_95th.py -c ``` ```bash python wan_util_95th.py -c -a MY-appliance-01 ``` ```bash python wan_util_95th.py -r ``` ```bash python wan_util_95th.py -c -d ``` -------------------------------- ### Example limit_appliances.json with appliance_subset Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/ec_telemetry_demo.html Example limit_appliances.json that limits telemetry collection to two appliances: REAL-HOSTNAME-01 and REAL-HOSTNAME-02. The file is copied into the ec-telemetry-discover container on build; to change it, edit the copy inside the container or modify the local file and rebuild with --build. ```json { "appliance_subset": [ "REAL-HOSTNAME-01", "REAL-HOSTNAME-02" ], "example": [ "Hostname-01", "SITE-HQ-02", "MyAppliance-A" ], "comment": "List of appliance hostnames to limit telemetry collection to" } ``` -------------------------------- ### Usage of get_api_key Source: https://pyedgeconnect.readthedocs.io/en/latest/release-notes/0.16.0-a1.html Example of calling the get_api_key function with a specific name parameter. ```python orch.get_api_key(name="myKey") ``` -------------------------------- ### Run preconfig.py with upload flag Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/generate_preconfig.html Executes the script to generate and upload valid preconfigs to the Orchestrator. ```bash python preconfig.py -c new_preconfigs.csv -u True ``` -------------------------------- ### Example log messages for EC-01 and EC-02 Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/ec_telemetry_demo.html Example log messages for two EdgeConnect appliances, EC-01 and EC-02, showing the log format: date/time, severity, container ID, hostname, IP, process timer, and message. The messages indicate failed login and failed ping with response code 256. ```plaintext 2022-09-12 22:03:33,033 - ERROR - cid: de11e3d0706e - EC-01 @ 10.41.2.125 - 2.8259: FAILED TO LOGIN 2022-09-12 22:03:34,271 - ERROR - cid: 1aa54222ea6b - EC-02 @ 10.41.0.70 - 4.0705: FAILED TO PING APPLIANCE -- ping response code: 256 ``` -------------------------------- ### Enable logging for Orchestrator and EdgeConnect Source: https://pyedgeconnect.readthedocs.io/en/latest/logging.html Demonstrates how to instantiate Orchestrator with console and success logging enabled, and EdgeConnect with file logging enabled. ```python orch_url = 'orchestrator.example.com' orch = Orchestrator(orch_url, log_console=True, log_success=True) # or ecos_url = '10.2.30.50' ec = EdgeConnect(ecos_url, log_file=True) ``` -------------------------------- ### Flake8 run command Source: https://pyedgeconnect.readthedocs.io/en/latest/code_formatting.html Run Flake8 against the pyedgeconnect modules or examples from the root repo directory. ```bash flake8 ``` -------------------------------- ### Conditional Logic in Jinja Template Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/generate_preconfig.html Example of using Jinja conditional operators to set values dynamically based on the appliance hostname. ```jinja2 {% if data['hostname'][-3:] == "-01" %} localMetric: 50 {% elif data['hostname'][-3:] == "-02" %} localMetric: 60 {% endif %} ``` -------------------------------- ### Initialize Edge Connect client with logging and SSL settings Source: https://pyedgeconnect.readthedocs.io/en/latest/_modules/pyedgeconnect.html Initializes the Edge Connect client with URL, API key, authentication mode, logging options, SSL verification, and timeout settings. Sets up request session, disables certificate warnings when verify_ssl is False, and configures file or console logging handlers. Prints a warning when log_success and log_file are both enabled. ```python self.url_prefix = "https://" + url + ":443/rest/json" self.timeout = timeout self.session = requests.Session() self.headers = {} # for API calls w/ just source as query param self.apiSrcId = "?source=menu_rest_apis_id" # for API calls w/ multiple query params self.apiSrcId2 = "&source=menu_rest_apis_id" # disable certificate warning messages / errors self.verify = verify_ssl if verify_ssl is False: requests.packages.urllib3.disable_warnings( category=InsecureRequestWarning ) # Allow logging successful response text/data if log_success # is set to True, warn user if logging to local file is also # enabled self.log_success = log_success if self.log_success and log_file: print( """ NOTE: Logging Response text from successful API calls has been enabled. Some responses can include sensitive data that you may not wish to retain in the log files. """ ) # Setup general log settings for messages and errors self.logger = logging.getLogger(f"ecos_{url}") self.formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) self.logger.setLevel(logging.INFO) # Setup logging file if log_file set to True if log_file: local_log_directory = "logging/" if not os.path.exists(local_log_directory): os.makedirs(local_log_directory) self.file_handler = logging.FileHandler( "{}sp_ecos.log".format(local_log_directory) ) self.file_handler.setFormatter(self.formatter) self.file_handler.setLevel(logging.INFO) self.logger.addHandler(self.file_handler) # Setup logging to console if log_console set to True if log_console: self.console_handler = logging.StreamHandler(sys.stdout) self.console_handler.setFormatter(self.formatter) self.console_handler.setLevel(logging.DEBUG) self.logger.setLevel(logging.DEBUG) self.logger.addHandler(self.console_handler) ``` -------------------------------- ### Run boost_timeseries.py for all appliances Source: https://pyedgeconnect.readthedocs.io/en/latest/examples/boost_timeseries.html Runs the boost_timeseries.py script to collect Boost timeseries data for all appliances. Requires the -c or -r argument at minimum, though this invocation does not include them. ```bash python boost_timeseries.py ```