Netbox GrapQL Inventory Task

GrapQL Sample Usage

NORFAB Netbox GrapQL Command Shell Reference

NorFab shell supports these command options for Netbox graphql task:

Python API Reference

Function to query Netbox v4 GraphQL API

Parameters:

Name	Type	Description	Default
instance	str	Netbox instance name	None
dry_run	bool	only return query content, do not run it	False
obj	dict	Object to query	None
filters	dict	Filters to apply to the query	None
fields	list	Fields to retrieve in the query	None
queries	dict	Dictionary of queries to execute	None
query_string	str	Raw query string to execute	None

Returns:

Type	Description
Union[dict, list]	GraphQL request data returned by Netbox

Raises:

Type	Description
RuntimeError	If required arguments are not provided
Exception	If GraphQL query fails

Source code in norfab\workers\netbox_worker.py

def graphql(
    self,
    instance: str = None,
    dry_run: bool = False,
    obj: dict = None,
    filters: dict = None,
    fields: list = None,
    queries: dict = None,
    query_string: str = None,
) -> Union[dict, list]:
    """
    Function to query Netbox v4 GraphQL API

    :param instance: Netbox instance name
    :param dry_run: only return query content, do not run it
    :param obj: Object to query
    :param filters: Filters to apply to the query
    :param fields: Fields to retrieve in the query
    :param queries: Dictionary of queries to execute
    :param query_string: Raw query string to execute
    :return: GraphQL request data returned by Netbox
    :raises RuntimeError: If required arguments are not provided
    :raises Exception: If GraphQL query fails
    """
    nb_params = self._get_instance_params(instance)
    ret = Result(task=f"{self.name}:graphql")

    # form graphql query(ies) payload
    if queries:
        queries_list = []
        for alias, query_data in queries.items():
            query_data["alias"] = alias
            if self.nb_version[0] == 4:
                queries_list.append(_form_query_v4(**query_data))
            elif self.nb_version[0] == 3:
                queries_list.append(_form_query_v3(**query_data))
        queries_strings = "    ".join(queries_list)
        query = f"query {{{queries_strings}}}"
    elif obj and filters and fields:
        if self.nb_version[0] == 4:
            query = _form_query_v4(obj, filters, fields)
        elif self.nb_version[0] == 3:
            query = _form_query_v3(obj, filters, fields)
        query = f"query {{{query}}}"
    elif query_string:
        query = query_string
    else:
        raise RuntimeError(
            f"{self.name} - graphql method expects quieries argument or obj, filters, "
            f"fields arguments or query_string argument provided"
        )
    payload = json.dumps({"query": query})

    # form and return dry run response
    if dry_run:
        ret.result = {
            "url": f"{nb_params['url']}/graphql/",
            "data": payload,
            "verify": nb_params.get("ssl_verify", True),
            "headers": {
                "Content-Type": "application/json",
                "Accept": "application/json",
                "Authorization": f"Token ...{nb_params['token'][-6:]}",
            },
        }
        return ret

    # send request to Netbox GraphQL API
    log.debug(
        f"{self.name} - sending GraphQL query '{payload}' to URL '{nb_params['url']}/graphql/'"
    )
    req = requests.post(
        url=f"{nb_params['url']}/graphql/",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Token {nb_params['token']}",
        },
        data=payload,
        verify=nb_params.get("ssl_verify", True),
        timeout=(self.netbox_connect_timeout, self.netbox_read_timeout),
    )
    try:
        req.raise_for_status()
    except Exception as e:
        raise Exception(
            f"{self.name} -  Netbox GraphQL query failed, query '{query}', "
            f"URL '{req.url}', status-code '{req.status_code}', reason '{req.reason}', "
            f"response content '{req.text}'"
        )

    # return results
    reply = req.json()
    if reply.get("errors"):
        msg = f"{self.name} - GrapQL query error '{reply['errors']}', query '{payload}'"
        log.error(msg)
        ret.errors.append(msg)
        if reply.get("data"):
            ret.result = reply["data"]  # at least return some data
    elif queries or query_string:
        ret.result = reply["data"]
    else:
        ret.result = reply["data"][obj]

    return ret