rest-connector

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}")