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 Netbox 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	Any	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\netbox_worker.py

@Task(
    input=RestInput,
    output=RestResult,
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Call NetBox REST API",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": False,
            "openWorldHint": True,
        }
    },
)
def rest(
    self,
    job: Job,
    instance: Union[None, str] = None,
    method: str = "get",
    api: str = "",
    **kwargs: Any,
) -> Result:
    """
    Sends a request to the Netbox REST API.

    Args:
        instance (str, optional): The Netbox 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)
    api = api.strip("/")

    log.info(f"{self.name} - REST {method.upper()} '{nb_params['url']}/api/{api}/'")

    # 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,
    )

    try:
        response.raise_for_status()
    except Exception as e:
        log.error(
            f"{self.name} - REST {method.upper()} '{nb_params['url']}/api/{api}/' failed, status {response.status_code}, error: {e}"
        )
        ret.result = response.status_code
        return ret

    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