pyATS REST Connector

Common errors

• requests.exceptions.ConnectionError: ('Connection aborted.', ConnectionRefusedError(111, 'Connection refused'))

  cause The target REST API server is not running, is unreachable, or the host/port configured in the testbed is incorrect. A firewall might also be blocking the connection.
  fix Verify the API endpoint's host and port in your testbed configuration, ensure the service is running, and check network connectivity (e.g., `ping`, `telnet host port`) and firewall rules.

• requests.exceptions.SSLError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self-signed certificate in certificate chain

  cause The SSL certificate of the target API server cannot be verified by the system's trusted certificate authorities. This often happens with self-signed certificates or internal CAs.
  fix For non-production development or internal systems, set `ssl_verify: False` in your testbed connection arguments (e.g., `arguments: {ssl_verify: False}`). For production, install the correct CA certificate chain on the system or provide its path via the `ca_bundle` argument.

• pyats.connections.rest.exceptions.RestConnectionError: API returned status code 401: Unauthorized

  cause The provided credentials (username/password/token) are incorrect, missing, or lack the necessary permissions to access the requested resource on the API.
  fix Double-check the `username` and `password` or other authentication details (e.g., API key in `headers`) in your testbed definition. Ensure environment variables like `REST_USERNAME` and `REST_PASSWORD` are correctly set if using them.

• json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

  cause The API response was not valid JSON or was empty, but the pyATS `rest-connector` implicitly tried to parse it as JSON (e.g., via `response.json()`). This can happen if the API returns plain text, HTML, or an error message that isn't JSON.
  fix Inspect the raw response using `response.text` to understand what the API is actually returning. Ensure the API endpoint indeed returns JSON for the specific request, or explicitly handle non-JSON responses before attempting to parse as JSON.

Warnings

• gotcha By default, `ssl_verify` is `True` for secure connections. For internal APIs with self-signed certificates, users often disable it (e.g., `ssl_verify: False` in testbed arguments). This should be strictly avoided in production environments due to significant security risks.
  Fix: For development, set `ssl_verify: False`. For production, ensure valid, trusted certificates are used, or explicitly trust the CA by providing a `ca_bundle` path in connection arguments.
• gotcha While direct instantiation of `pyats.connections.rest.Rest` is possible, the recommended and most robust way to use `rest-connector` within the pyATS framework is through a `Testbed` definition. This approach handles connection lifecycle, integrates with other pyATS features, and centralizes configuration.
  Fix: Always define REST devices within a `Testbed` YAML file and load it using `from pyats.topology import Testbed`.
• gotcha REST APIs often impose rate limits or require pagination for large datasets. `rest-connector` provides basic request functionality; users are responsible for implementing logic to handle these API-specific constraints (e.g., retries, sleep intervals, parsing pagination headers/links).
  Fix: Implement retry mechanisms (e.g., using `tenacity`), check `Retry-After` headers for rate limits, and write explicit loops or functions to handle paginated responses based on API documentation.
• breaking Major version upgrades of the core `pyATS` framework (e.g., from 21.x to 22.x) can sometimes introduce changes to connection arguments, credential handling, or underlying library versions that affect `rest-connector` behavior. Always consult the pyATS release notes.
  Fix: Review the pyATS release notes for any breaking changes related to connection plugins. Update testbed configurations and Python code to align with the new pyATS version's requirements.

Install

• pip install rest-connector pyats Install pyATS and rest-connector

Imports

• Testbed
  wrong:

  from rest_connector.topology import Testbed

  correct:

  from pyats.topology import Testbed

  The core Testbed class is part of pyATS topology, not directly from rest-connector.
• Rest
  wrong:

  from rest_connector import Rest

  correct:

  from pyats.connections.rest import Rest

  The Rest connection class is located within the pyATS connections namespace, not at the top level of the rest-connector package.

Quickstart

This quickstart demonstrates how to set up a REST API connection using a pyATS testbed configuration. It involves defining a device of `os: rest` type in a YAML testbed file, then connecting to it from Python and making a simple GET request. Credentials are securely fetched from environment variables. A mock setup is provided for immediate execution without a live API or testbed file.

import os
from pyats.topology import Testbed

# 1. Create a testbed.yaml file:
#   devices:
#     my_rest_api:
#       os: rest
#       connections:
#         rest:
#           class: pyats.connections.rest.Rest
#           arguments:
#             host: example.com
#             port: 443
#             ssl_verify: True
#             headers:
#               Content-Type: application/json
#             credentials:
#               username: ${{ REST_USERNAME }}
#               password: ${{ REST_PASSWORD }}

# Set environment variables for credentials (for a real API)
# os.environ['REST_USERNAME'] = 'myuser'
# os.environ['REST_PASSWORD'] = 'mypassword'

# For this example, we'll use a placeholder and mock if needed
# In a real scenario, example.com/api/data would be queried.
# For a runnable example that doesn't hit a real server, we assume a simple GET
# and print the raw text, as full mocking is out of quickstart scope.

# Mocking a testbed for a runnable example without actual file I/O
# In a real scenario, this would load from a 'testbed.yaml' file.
class MockConnection:
    def __init__(self, host, port, ssl_verify, headers, credentials):
        self.host = host
        self.port = port
        self.ssl_verify = ssl_verify
        self.headers = headers
        self.credentials = credentials

    def get(self, path, headers=None, verify=None, auth=None):
        class MockResponse:
            def __init__(self, text, status_code=200):
                self.text = text
                self.status_code = status_code

            def json(self):
                import json
                return json.loads(self.text)

        if path == '/api/data':
            print(f"[Mock] GET request to {self.host}:{self.port}{path}")
            return MockResponse('{"status": "success", "data": [1,2,3]}')
        return MockResponse('{"error": "not found"}', status_code=404)

class MockRestDevice:
    def __init__(self, name, connections_config):
        self.name = name
        self.rest = MockConnection(
            connections_config['rest']['arguments']['host'],
            connections_config['rest']['arguments']['port'],
            connections_config['rest']['arguments']['ssl_verify'],
            connections_config['rest']['arguments']['headers'],
            connections_config['rest']['arguments']['credentials']
        )

    def connect(self):
        print(f"[Mock] Connecting to device {self.name}")
        # Simulate connection logic

# Simulate a simple testbed and device using the mock classes
testbed_config = {
    'devices': {
        'my_rest_api': {
            'os': 'rest',
            'connections': {
                'rest': {
                    'class': 'pyats.connections.rest.Rest',
                    'arguments': {
                        'host': 'example.com',
                        'port': 443,
                        'ssl_verify': True,
                        'headers': {'Content-Type': 'application/json'},
                        'credentials': {
                            'username': os.environ.get('REST_USERNAME', 'mockuser'),
                            'password': os.environ.get('REST_PASSWORD', 'mockpass')
                        }
                    }
                }
            }
        }
    }
}

# In a real scenario, load from file:
# testbed = Testbed('testbed.yaml')
# For this quickstart, we'll create a mock Testbed/device
class MockTestbed:
    def __init__(self, config):
        self.devices = {
            name: MockRestDevice(name, device_config)
            for name, device_config in config['devices'].items()
        }

mock_testbed = MockTestbed(testbed_config)
device = mock_testbed.devices['my_rest_api']

# 2. Connect to the device (initiates the REST session)
device.connect()

# 3. Make an API call
try:
    response = device.rest.get('/api/data')
    print(f"API Response Status: {response.status_code}")
    if response.status_code == 200:
        print(f"API Response JSON: {response.json()}")
    else:
        print(f"API Error: {response.text}")
except Exception as e:
    print(f"An error occurred: {e}")