Nornir Service Parse Tasks

task api names: parse_napalm, parse_ttp, parse_textfsm

Three tasks for parsing network device CLI output into structured data:

• parse_napalm — retrieves structured data via NAPALM getters (e.g. get_facts, get_interfaces). No template required.
• parse_ttp — collects CLI output and parses it with a TTP template. Supports inline templates, ttp:// paths from ttp_templates, nf:// file paths, and HTTP URLs. Template input definitions drive which commands are collected automatically.
• parse_textfsm — collects CLI output and parses it with a TextFSM template. When no template is provided, NTC-Templates auto-detection is used based on device platform and command.

NORFAB Shell Reference

NorFab shell commands for Nornir parse tasks:

nf#man tree nornir.parse

R - required field, M - supports multiline input, D - dynamic key

root
└── nornir:    Nornir service
    └── parse:    Parse network devices output
        ├── napalm:    Parse devices output using NAPALM getters
        │   ├── timeout:    Job timeout
        │   ├── workers:    Filter workers to target, default 'all'
        │   ├── verbose-result:    Control output details, default 'False'
        │   ├── progress:    Display progress events, default 'True'
        │   ├── nowait:    Do not wait for job to complete, default 'False'
        │   ├── add-details:    Add task details to results, default 'False'
        │   ├── num-workers:    RetryRunner number of threads for tasks execution
        │   ├── num-connectors:    RetryRunner number of threads for device connections
        │   ├── connect-retry:    RetryRunner number of connection attempts
        │   ├── task-retry:    RetryRunner number of attempts to run task
        │   ├── reconnect-on-fail:    RetryRunner perform reconnect to host on task failure
        │   ├── connect-check:    RetryRunner test TCP connection before opening actual connection
        │   ├── connect-timeout:    RetryRunner timeout in seconds to wait for test TCP connection to establish
        │   ├── creds-retry:    RetryRunner list of connection credentials and parameters to retry
        │   ├── tf:    File group name to save task results to on worker file system
        │   ├── tf-skip-failed:    Save results to file for failed tasks
        │   ├── diff:    File group name to run the diff for
        │   ├── diff-last:    File version number to diff, default is 1 (last)
        │   ├── FO:    Filter hosts using Filter Object
        │   ├── FB:    Filter hosts by name using Glob Patterns
        │   ├── FH:    Filter hosts by hostname
        │   ├── FC:    Filter hosts containment of pattern in name
        │   ├── FR:    Filter hosts by name using Regular Expressions
        │   ├── FG:    Filter hosts by group
        │   ├── FP:    Filter hosts by hostname using IP Prefix
        │   ├── FL:    Filter hosts by names list
        │   ├── FM:    Filter hosts by platform
        │   ├── FX:    Filter hosts excluding them by name
        │   ├── FN:    Negate the match
        │   └── getters (R):    Select NAPALM getters
        ├── ttp:    Parse devices output using TTP templates
        │   ├── timeout:    Job timeout
        │   ├── workers:    Filter workers to target, default 'all'
        │   ├── verbose-result:    Control output details, default 'False'
        │   ├── progress:    Display progress events, default 'True'
        │   ├── nowait:    Do not wait for job to complete, default 'False'
        │   ├── add-details:    Add task details to results, default 'False'
        │   ├── num-workers:    RetryRunner number of threads for tasks execution
        │   ├── num-connectors:    RetryRunner number of threads for device connections
        │   ├── connect-retry:    RetryRunner number of connection attempts
        │   ├── task-retry:    RetryRunner number of attempts to run task
        │   ├── reconnect-on-fail:    RetryRunner perform reconnect to host on task failure
        │   ├── connect-check:    RetryRunner test TCP connection before opening actual connection
        │   ├── connect-timeout:    RetryRunner timeout in seconds to wait for test TCP connection to establish
        │   ├── creds-retry:    RetryRunner list of connection credentials and parameters to retry
        │   ├── tf:    File group name to save task results to on worker file system
        │   ├── tf-skip-failed:    Save results to file for failed tasks
        │   ├── diff:    File group name to run the diff for
        │   ├── diff-last:    File version number to diff, default is 1 (last)
        │   ├── FO:    Filter hosts using Filter Object
        │   ├── FB:    Filter hosts by name using Glob Patterns
        │   ├── FH:    Filter hosts by hostname
        │   ├── FC:    Filter hosts containment of pattern in name
        │   ├── FR:    Filter hosts by name using Regular Expressions
        │   ├── FG:    Filter hosts by group
        │   ├── FP:    Filter hosts by hostname using IP Prefix
        │   ├── FL:    Filter hosts by names list
        │   ├── FM:    Filter hosts by platform
        │   ├── FX:    Filter hosts excluding them by name
        │   ├── FN:    Negate the match
        │   ├── template:    TTP Template to parse commands output, supports ttp:// path
        │   ├── get:    Getter TTP Template name to use
        │   ├── commands (M):    List of commands to collect form devices
        │   ├── plugin:    CLI connection plugin parameters
        │   │   ├── netmiko:    Use Netmiko plugin to collect output from devices
        │   │   │   ├── enable:    Attempt to enter enable-mode
        │   │   │   ├── use-timing:    switch to send command timing method
        │   │   │   ├── expect-string:    Regular expression pattern to use for determining end of output
        │   │   │   ├── read-timeout:    Maximum time to wait looking for pattern
        │   │   │   ├── auto-find-prompt:    Use find_prompt() to override base prompt
        │   │   │   ├── strip-prompt:    Remove the trailing router prompt from the output
        │   │   │   ├── strip-command:    Remove the echo of the command from the output
        │   │   │   ├── normalize:    Ensure the proper enter is sent at end of command
        │   │   │   ├── use-textfsm:    Process command output through TextFSM template
        │   │   │   ├── textfsm-template:    Name of template to parse output with
        │   │   │   ├── use-ttp:    Process command output through TTP template
        │   │   │   ├── ttp-template:    Name of template to parse output with
        │   │   │   ├── use-genie:    Process command output through PyATS/Genie parser
        │   │   │   ├── cmd-verify:    Verify command echo before proceeding
        │   │   │   ├── interval:    Interval between sending commands
        │   │   │   ├── use-ps:    Use send command promptless method
        │   │   │   ├── use-ps-timeout:    Promptless mode absolute timeout
        │   │   │   ├── split-lines:    Split multiline string to individual commands
        │   │   │   ├── new-line-char:    Character to replace with new line before sending to device, default is _br_
        │   │   │   ├── repeat:    Number of times to repeat the commands
        │   │   │   ├── stop-pattern:    Stop commands repeat if output matches provided glob pattern
        │   │   │   ├── repeat-interval:    Time in seconds to wait between repeating commands
        │   │   │   └── return-last:    Returns requested last number of commands outputs
        │   │   ├── scrapli:    Use Scrapli plugin to collect output from devices
        │   │   │   ├── strip-prompt:    Strip prompt from returned output
        │   │   │   ├── failed-when-contains:    String or list of strings indicating failure if found in response
        │   │   │   ├── timeout-ops:    Timeout ops value for this operation
        │   │   │   ├── interval:    Interval between sending commands
        │   │   │   ├── split-lines:    Split multiline string to individual commands
        │   │   │   ├── new-line-char:    Character to replace with new line before sending to device, default is _br_
        │   │   │   ├── repeat:    Number of times to repeat the commands
        │   │   │   ├── stop-pattern:    Stop commands repeat if output matches provided glob pattern
        │   │   │   ├── repeat-interval:    Time in seconds to wait between repeating commands
        │   │   │   └── return-last:    Returns requested last number of commands outputs
        │   │   └── napalm:    Use NAPALM plugin to collect output from devices
        │   │       ├── interval:    Interval between sending commands
        │   │       ├── split-lines:    Split multiline string to individual commands
        │   │       └── new-line-char:    Character to replace with new line before sending to device, default is _br_
        │   ├── enable:    Enter exec mode
        │   └── structure:    TTP Results structure
        └── textfsm:    Parse devices output using TextFSM templates
            ├── timeout:    Job timeout
            ├── workers:    Filter workers to target, default 'all'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── nowait:    Do not wait for job to complete, default 'False'
            ├── add-details:    Add task details to results, default 'False'
            ├── num-workers:    RetryRunner number of threads for tasks execution
            ├── num-connectors:    RetryRunner number of threads for device connections
            ├── connect-retry:    RetryRunner number of connection attempts
            ├── task-retry:    RetryRunner number of attempts to run task
            ├── reconnect-on-fail:    RetryRunner perform reconnect to host on task failure
            ├── connect-check:    RetryRunner test TCP connection before opening actual connection
            ├── connect-timeout:    RetryRunner timeout in seconds to wait for test TCP connection to establish
            ├── creds-retry:    RetryRunner list of connection credentials and parameters to retry
            ├── tf:    File group name to save task results to on worker file system
            ├── tf-skip-failed:    Save results to file for failed tasks
            ├── diff:    File group name to run the diff for
            ├── diff-last:    File version number to diff, default is 1 (last)
            ├── FO:    Filter hosts using Filter Object
            ├── FB:    Filter hosts by name using Glob Patterns
            ├── FH:    Filter hosts by hostname
            ├── FC:    Filter hosts containment of pattern in name
            ├── FR:    Filter hosts by name using Regular Expressions
            ├── FG:    Filter hosts by group
            ├── FP:    Filter hosts by hostname using IP Prefix
            ├── FL:    Filter hosts by names list
            ├── FM:    Filter hosts by platform
            ├── FX:    Filter hosts excluding them by name
            ├── FN:    Negate the match
            ├── template:    Path to a TextFSM template file
            └── commands (M):    List of commands to parse form devices
nf#

Python API Reference

Parse network device output using NAPALM getters.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
getters	Union[str, list[str]]	NAPALM getter name(s) to call, e.g. get_facts.	'get_facts'
to_dict	bool	If True, return results as a dictionary keyed by hostname.	True
add_details	bool	If True, include extra task details in the result.	False
**kwargs	Any	Additional arguments forwarded to Nornir host filtering.	{}

Returns:

Type	Description
Result	Result containing parsed NAPALM getter data.

Source code in norfab\workers\nornir_worker\parse_task.py

@Task(fastapi={"methods": ["POST"]})
def parse_napalm(
    self,
    job: Job,
    getters: Union[str, list[str]] = "get_facts",
    to_dict: bool = True,
    add_details: bool = False,
    **kwargs: Any,
) -> Result:
    """
    Parse network device output using NAPALM getters.

    Args:
        job: NorFab Job object containing relevant metadata.
        getters: NAPALM getter name(s) to call, e.g. ``get_facts``.
        to_dict: If True, return results as a dictionary keyed by hostname.
        add_details: If True, include extra task details in the result.
        **kwargs: Additional arguments forwarded to Nornir host filtering.

    Returns:
        Result containing parsed NAPALM getter data.
    """
    ret = Result(task=f"{self.name}:parse_napalm", result={} if to_dict else [])

    filtered_nornir, ret = self.filter_hosts_and_validate(kwargs, ret)
    if ret.status == "no_match":
        return ret

    nr = self._add_processors(filtered_nornir, kwargs, job)  # add processors
    result = nr.run(task=napalm_get, getters=getters, **kwargs)
    ret.result = ResultSerializer(result, to_dict=to_dict, add_details=add_details)
    ret.failed = result.failed  # failed is true if any of the hosts failed

    return ret

Parse network device output using a TTP template.

Fetches CLI output from devices based on the template's input definitions, then parses the collected output with TTP.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
template	Union[str, None]	TTP template string, URL, or ttp:// path.	None
strict	bool	If True, raise RuntimeError when any host yields no parsed output.	True
structure	str	TTP result structure one of flat_list, list, or dictionary.	'flat_list'
commands	Union[str, list[str], None]	Fallback commands to collect when the template defines no explicit input commands.	None
get	Union[str, None]	TTP getter template name to use.	None
plugin	str	Nornir connection plugin for CLI collection (default netmiko).	'netmiko'
**kwargs	Any	Additional arguments forwarded to Nornir host filtering and the CLI collection task.	{}

Returns:

Type	Description
Result	Result containing per-host TTP parsing output.

Raises:

Type	Description
RuntimeError	If strict=True and a host produces empty results.

Source code in norfab\workers\nornir_worker\parse_task.py

@Task(fastapi={"methods": ["POST"]}, input=ParseTTPInput)
def parse_ttp(
    self,
    job: Job,
    template: Union[str, None] = None,
    strict: bool = True,
    structure: str = "flat_list",
    commands: Union[str, list[str], None] = None,
    get: Union[str, None] = None,
    plugin: str = "netmiko",
    **kwargs: Any,
) -> Result:
    """
    Parse network device output using a TTP template.

    Fetches CLI output from devices based on the template's input
    definitions, then parses the collected output with TTP.

    Args:
        job: NorFab Job object containing relevant metadata.
        template: TTP template string, URL, or ``ttp://`` path.
        strict: If True, raise ``RuntimeError`` when any host yields no parsed output.
        structure: TTP result structure one of ``flat_list``, ``list``, or ``dictionary``.
        commands: Fallback commands to collect when the template defines no explicit input commands.
        get: TTP getter template name to use.
        plugin: Nornir connection plugin for CLI collection (default ``netmiko``).
        **kwargs: Additional arguments forwarded to Nornir host filtering and the CLI collection task.

    Returns:
        Result containing per-host TTP parsing output.

    Raises:
        RuntimeError: If ``strict=True`` and a host produces empty results.
    """
    kwargs["to_dict"] = True
    kwargs["add_details"] = True
    filters = {k: kwargs[k] for k in kwargs.keys() if k in FFun_functions}
    ret = Result(task=f"{self.name}:parse_ttp", result={})

    # check has hosts to run for
    filtered_nornir, ret = self.filter_hosts_and_validate(kwargs, ret)
    if ret.status == "no_match":
        return ret

    # download TTP template
    if self.is_url(template):
        template = self.fetch_file(template, raise_on_fail=True, read=True)
    elif template and template.startswith("ttp://"):
        template = get_template(path=template)
    elif get:
        template = get_template(get=get)

    # go over template's inputs and collect commands to fetch from devices
    cli_runs = []
    parser = ttp(template=template, log_level="ERROR")
    for template_name, inputs in parser.get_input_load().items():
        inputs = inputs or {"Default_Input": {}}
        for input_name, input_params in inputs.items():
            cli_runs.append(
                {
                    "commands": input_params.get("commands") or commands,
                    "input_name": input_name,
                    "template_name": template_name,
                    "params": input_params,
                }
            )

    job.event(
        f"collecting output from {len(filtered_nornir.inventory.hosts)} devices for {len(cli_runs)} TTP template input(s)"
    )

    # collect commands output from devices
    for cli_run in cli_runs:
        cli_run_filters = {
            k: cli_run["params"][k]
            for k in cli_run["params"].keys()
            if k in FFun_functions
        }
        cli_run_kwargs = {
            **filters,
            **kwargs,
            **cli_run_filters,
        }
        if cli_run["params"].get("platform"):
            cli_run_kwargs["FM"] = cli_run["params"]["platform"]
        result = self.cli(
            job=job, commands=cli_run["commands"], **cli_run_kwargs, plugin=plugin
        )
        if result.failed:
            ret.failed = True
            log.error(f"Failed collecting commands output, errors: {result.errors}")
            job.event("failed collecting commands output", severity="ERROR")
            ret.errors.extend(result.errors)
            ret.messages.extend(result.messages)
            continue
        # parse commands output for each host
        for hname, hres in result.result.items():
            input_data = []
            for cmdname, cmdres in hres.items():
                if cmdres["failed"]:
                    msg = f"Failed collecting command '{cmdname}' output from '{hname}', error: {cmdres['exception']}"
                    log.error(msg)
                    ret.errors.append(msg)
                    job.event(msg, severity="ERROR")
                    continue
                input_data.append(cmdres["result"])
            # parse command results
            if input_data:
                parser = ttp(template=template, log_level="ERROR")
                parser.add_input(
                    data="\n\n".join(input_data),
                    input_name=cli_run["input_name"],
                    template_name=cli_run["template_name"],
                )
                # run parsing in single process
                parser.parse(one=True)
                ret.result[hname] = parser.result(
                    structure=structure, templates=cli_run["template_name"]
                )
            else:
                msg = f"No input data collected for '{hname}' device"
                log.error(msg)
                ret.errors.append(msg)
                ret.result[hname] = None
                job.event(msg, severity="ERROR")

    # if strict - check hosts parsing results and raise error if empty
    if strict:
        for hname, hres in ret.result.items():
            if not hres or hres in [[{}], [], {}, [[]]]:
                raise RuntimeError(f"{hname} host parsing produced no results")

    return ret

Parse network device output using TextFSM templates.

Collects CLI output from devices and parses it using a TextFSM template. When no template is provided, NTC-Templates auto-detection is used based on the device platform and command.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
commands	Union[str, list[str], None]	Command(s) to collect from devices.	None
template	Union[str, None]	Path to a TextFSM template file, or nf:// URL to a file stored in NorFab file sharing. If omitted, NTC-Templates auto-detection is used.	None
to_dict	bool	If True, return results as a dictionary keyed by hostname.	True
add_details	bool	If True, include extra task details in the result.	False
**kwargs	Any	Additional arguments forwarded to Nornir host filtering and the CLI collection task.	{}

Returns:

Type	Description
Result	Result containing per-host TextFSM parsed output.

Source code in norfab\workers\nornir_worker\parse_task.py

@Task(fastapi={"methods": ["POST"]})
def parse_textfsm(
    self,
    job: Job,
    commands: Union[str, list[str], None] = None,
    template: Union[str, None] = None,
    to_dict: bool = True,
    add_details: bool = False,
    **kwargs: Any,
) -> Result:
    """
    Parse network device output using TextFSM templates.

    Collects CLI output from devices and parses it using a TextFSM template.
    When no template is provided, NTC-Templates auto-detection is used based
    on the device platform and command.

    Args:
        job: NorFab Job object containing relevant metadata.
        commands: Command(s) to collect from devices.
        template: Path to a TextFSM template file, or ``nf://`` URL to a
            file stored in NorFab file sharing. If omitted, NTC-Templates
            auto-detection is used.
        to_dict: If True, return results as a dictionary keyed by hostname.
        add_details: If True, include extra task details in the result.
        **kwargs: Additional arguments forwarded to Nornir host filtering
            and the CLI collection task.

    Returns:
        Result containing per-host TextFSM parsed output.
    """
    ret = Result(task=f"{self.name}:parse_textfsm", result={} if to_dict else [])

    filtered_nornir, ret = self.filter_hosts_and_validate(kwargs, ret)
    if ret.status == "no_match":
        return ret

    # resolve template from URL or local nf:// path
    if template and self.is_url(template):
        template = self.fetch_file(template, raise_on_fail=True, read=False)

    result = self.cli(
        job=job,
        commands=commands,
        plugin="netmiko",
        to_dict=to_dict,
        add_details=add_details,
        use_textfsm=True,
        textfsm_template=template,
        **kwargs,
    )
    ret.result = result.result
    ret.failed = result.failed
    ret.errors = result.errors

    return ret