Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

http requests improvements for get_device_config #51

Merged
merged 6 commits into from
Jun 24, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .pylintrc
Original file line number Diff line number Diff line change
@@ -1,2 +1,5 @@
[MASTER]
disable=C0103,C0301,R0903,C0116,W0603,W0718
[DESIGN]
# needed to match number of arguments in requests.send
max-args=7
2 changes: 2 additions & 0 deletions nrx.conf
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ NB_API_URL = 'https://demo.netbox.dev'
NB_API_TOKEN = ''
# Peform TLS certification validation
TLS_VALIDATE = true
# API request timeout, in seconds
API_TIMEOUT = 10
# Output format to use for export: 'gml' | 'cyjs' | 'clab'. Alternatively, use --output argument
OUTPUT_FORMAT = 'clab'
# List of NetBox Device Roles to export
Expand Down
40 changes: 30 additions & 10 deletions nrx/nrx.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,8 @@
import toml
import pynetbox
import requests
from requests.adapters import HTTPAdapter
from requests.exceptions import RequestException, Timeout, HTTPError
import urllib3
import networkx as nx
import jinja2
Expand Down Expand Up @@ -69,6 +71,17 @@ def error_debug(err, d):
debug(d)
error(err)

class TimeoutHTTPAdapter(HTTPAdapter):
"""HTTPAdapter with custom API timeout"""
def __init__(self, timeout, *args, **kwargs):
self.timeout = timeout
super().__init__(*args, **kwargs)

def send(self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None):
if timeout is None:
timeout = self.timeout
return super().send(request, stream, timeout, verify, cert, proxies)

class NBNetwork:
"""Class to hold network topology data exported from NetBox"""
def __init__(self):
Expand Down Expand Up @@ -98,6 +111,10 @@ def __init__(self, config):
if not config['tls_validate']:
self.nb_session.http_session.verify = False
urllib3.disable_warnings()
if config['api_timeout'] > 0:
adapter = TimeoutHTTPAdapter(config['api_timeout'])
self.nb_session.http_session.mount("http://", adapter)
self.nb_session.http_session.mount("https://", adapter)
print(f"Connecting to NetBox at: {config['nb_api_url']}")
if len(config['export_site']) > 0:
debug(f"Fetching site: {config['export_site']}")
Expand Down Expand Up @@ -218,16 +235,18 @@ def _get_device_config(self, device):
'Accept': 'application/json'
}
url = f"{self.config['nb_api_url']}/api/dcim/devices/{device.id}/render-config/"
response = requests.post(url, headers=headers)
if response.status_code == 200:
try:
config_response = ast.literal_eval(response.text)
if "content" in config_response:
return config_response["content"]
except (SyntaxError) as e:
debug("Get device configuration failed: can't parse rendered configuration")
else:
debug(f"Get device configuration request failed with status code: {response.status_code}")
try:
response = requests.post(url, headers=headers, timeout=self.config['api_timeout'], verify=self.config['tls_validate'])
response.raise_for_status() # Raises an HTTPError if the response status is an error
config_response = ast.literal_eval(response.text)
if "content" in config_response:
return config_response["content"]
except HTTPError as e:
debug(f"{device.name}: Get device configuration request failed: {e}")
except (Timeout, RequestException) as e:
debug(f"{device.name}: Get device configuration failed: {e}")
except SyntaxError as e:
debug(f"{device.name}: Get device configuration failed: can't parse rendered configuration - {e}")
return ""

def _trace_cable(self, cable):
Expand Down Expand Up @@ -697,6 +716,7 @@ def load_toml_config(filename):
'nb_api_url': '',
'nb_api_token': '',
'tls_validate': True,
'api_timeout': 10,
'output_format': 'cyjs',
'export_device_roles': ["router", "core-switch", "access-switch", "distribution-switch", "tor-switch"],
'device_role_levels': {
Expand Down