Netbox GrapQL Inventory Task

task api name: graphql

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 v3 or Netbox v4 GraphQL API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	Union[None, str]	Netbox instance name	None
dry_run	bool	only return query content, do not run it	False
obj	Union[str, dict]	Object to query	None
filters	Union[None, dict, str]	Filters to apply to the query	None
fields	Union[None, list]	Fields to retrieve in the query	None
queries	Union[None, dict]	Dictionary of queries to execute	None
query_string	str	Raw query string to execute	None

Returns:

Name	Type	Description
dict	Result	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

@Task(fastapi={"methods": ["POST"]})
def graphql(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    obj: Union[str, dict] = None,
    filters: Union[None, dict, str] = None,
    fields: Union[None, list] = None,
    queries: Union[None, dict] = None,
    query_string: str = None,
) -> Result:
    """
    Function to query Netbox v3 or Netbox v4 GraphQL API.

    Args:
        job: NorFab Job object containing relevant metadata
        instance: Netbox instance name
        dry_run: only return query content, do not run it
        obj: Object to query
        filters: Filters to apply to the query
        fields: Fields to retrieve in the query
        queries: Dictionary of queries to execute
        query_string: Raw query string to execute

    Returns:
        dict: GraphQL request data returned by Netbox

    Raises:
        RuntimeError: If required arguments are not provided
        Exception: If GraphQL query fails
    """
    nb_params = self._get_instance_params(instance)
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:graphql", resources=[instance])

    # form graphql query(ies) payload
    if queries:
        queries_list = []
        for alias, query_data in queries.items():
            query_data["alias"] = alias
            if self.nb_version[instance][0] == 4:
                queries_list.append(_form_query_v4(**query_data))
            elif self.nb_version[instance][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[instance][0] == 4:
            query = _form_query_v4(obj, filters, fields)
        elif self.nb_version[instance][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