Skip to content

Netbox Rest Inventory Task¤

task api name: rest

Rest Task Sample Usage¤

NORFAB Netbox GrapQL Command Shell Reference¤

NorFab shell supports these command options for Netbox rest task:

Python API Reference¤

Sends a request to the Netbox REST API.

Parameters:

Name Type Description Default
instance str

The instance name to get parameters for.

None
method str

The HTTP method to use for the request (e.g., 'get', 'post'). Defaults to "get".

'get'
api str

The API endpoint to send the request to. Defaults to "".

''
**kwargs

Additional arguments to pass to the request (e.g., params, data, json).

{}

Returns:

Type Description
Result

Union[dict, list]: The JSON response from the API, parsed into a dictionary or list.

Raises:

Type Description
HTTPError

If the HTTP request returned an unsuccessful status code.

Source code in norfab\workers\netbox_worker.py
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
@Task()
def rest(
    self,
    job: Job,
    instance: Union[None, str] = None,
    method: str = "get",
    api: str = "",
    **kwargs,
) -> Result:
    """
    Sends a request to the Netbox REST API.

    Args:
        instance (str, optional): The instance name to get parameters for.
        method (str, optional): The HTTP method to use for the request (e.g., 'get', 'post'). Defaults to "get".
        api (str, optional): The API endpoint to send the request to. Defaults to "".
        **kwargs: Additional arguments to pass to the request (e.g., params, data, json).

    Returns:
        Union[dict, list]: The JSON response from the API, parsed into a dictionary or list.

    Raises:
        requests.exceptions.HTTPError: If the HTTP request returned an unsuccessful status code.
    """
    ret = Result(task=f"{self.name}:rest", result={})
    nb_params = self._get_instance_params(instance)

    # send request to Netbox REST API
    response = getattr(requests, method)(
        url=f"{nb_params['url']}/api/{api}/",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Token {nb_params['token']}",
        },
        verify=nb_params.get("ssl_verify", True),
        **kwargs,
    )

    response.raise_for_status()
    try:
        ret.result = response.json()
    except Exception as e:
        log.debug(f"Failed to decode json, error: {e}")
        ret.result = response.text if response.text else response.status_code

    return ret