Add support for custom headers in all Netbox modules

changelogs/fragments/1433-add-custom-headers.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - Add support for custom headers for all modules

docs/getting_started/contributing/modules/architecture.rst

@@ -284,10 +284,11 @@ We set several instance attributes that are used within other methods throughout
       url = self.module.params["netbox_url"]
       token = self.module.params["netbox_token"]
       ssl_verify = self.module.params["validate_certs"]
+      headers = self.module.params["headers"]
 
       # Attempt to initiate connection to NetBox
       if nb_client is None:
-          self.nb = self._connect_netbox_api(url, token, ssl_verify)
+          self.nb = self._connect_netbox_api(url, token, ssl_verify, headers)
       else:
           self.nb = nb_client
           try:

plugins/doc_fragments/common.py

@@ -50,4 +50,8 @@ class ModuleDocFragment(object):
       - Certificate path
     required: false
     type: raw
+  headers:
+    description: Dictionary of headers to be passed to the NetBox API.
+    required: false
+    type: dict
 """

plugins/module_utils/netbox_utils.py

@@ -723,6 +723,7 @@
     query_params=dict(required=False, type="list", elements="str"),
     validate_certs=dict(type="raw", default=True),
     cert=dict(type="raw", required=False),
+    headers=dict(type="dict", required=False),
 )
 
 
@@ -751,10 +752,11 @@ def __init__(self, module, endpoint, nb_client=None):
         token = self.module.params["netbox_token"]
         ssl_verify = self.module.params["validate_certs"]
         cert = self.module.params["cert"]
+        headers = self.module.params["headers"]
 
         # Attempt to initiate connection to NetBox
         if nb_client is None:
-            self.nb = self._connect_netbox_api(url, token, ssl_verify, cert)
+            self.nb = self._connect_netbox_api(url, token, ssl_verify, cert, headers)
         else:
             self.nb = nb_client
             try:
@@ -804,10 +806,14 @@ def _version_check_greater(self, greater, lesser, greater_or_equal=False):
 
         return False
 
-    def _connect_netbox_api(self, url, token, ssl_verify, cert):
+    def _connect_netbox_api(self, url, token, ssl_verify, cert, headers=None):
         try:
             session = requests.Session()
             session.verify = ssl_verify
+            if isinstance(headers, str):
+                headers = json.load(headers)
+            if isinstance(headers, dict):
+                session.headers.update(headers)
             if cert:
                 session.cert = tuple(i for i in cert)
             nb = pynetbox.api(url, token=token)

tests/unit/module_utils/test_netbox_base_class.py

@@ -63,6 +63,7 @@ def fixture_arg_spec():
         "state": "present",
         "validate_certs": False,
         "cert": None,
+        "headers": None,
     }