NFAPI Reference

Environment file loading

NorFab loads a .env file before it initializes NorFabInventory. By default, it checks the directory containing inventory.yaml. When using inventory_data, it checks base_dir, or the current directory when base_dir is omitted. This makes the values available to inventory Jinja2 expressions and to NFCLI, broker, and worker processes.

The python-dotenv loader supports Bash-like assignments, comments, quoted and multiline values, the optional export directive, and ${VAR} expansion. It replaces variables already present in the process environment by default, so the local .env file takes precedence over shell and CI values.

nf = NorFab(inventory="./inventory.yaml")  # Loads ./.env when present.
environment = nf.list_environment_variables()

Set load_env_override=False to preserve existing process variables instead:

nf = NorFab(inventory="./inventory.yaml", load_env_override=False)

Python API

NorFab is a class that provides an interface for interacting with the NorFab system.

Attributes:

Name	Type	Description
client	NFPClient	The client instance for interfacing with the broker.
broker	Process	The process instance for the broker.
inventory	NorFabInventory	The inventory instance containing configuration data.
workers_processes	dict	A dictionary mapping worker names to their process instances and initialization events.
worker_plugins	dict	A dictionary mapping service names to their worker plugins.

Parameters:

Name	Type	Description	Default
inventory	str	OS path to NorFab inventory YAML file	'./inventory.yaml'
inventory_data	dict	dictionary with NorFab inventory	None
base_dir	str	OS path to base directory to anchor NorFab at	None
log_level	str	one or supported logging levels - CRITICAL, ERROR, WARNING, INFO, DEBUG	None
load_env_override	bool	whether .env values override existing environment variables	True
configure_logging	bool	configure NorFab process logging during initialization	False
logging_name	str	process identity to use in the NFAPI JSONL log filename	'nfapi'

Example:

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
nf.start(run_broker=True, run_workers=["my-worker-1"])
NFCLIENT = nf.make_client()

Example using dictionary inventory data:

from norfab.core.nfapi import NorFab

data = {
    'broker': {'endpoint': 'tcp://127.0.0.1:5555'},
    'workers': {'my-worker-1': ['workers/common.yaml'],
}

nf = NorFab(inventory_data=data, base_dir="./")
nf.start(run_broker=True, run_workers=["my-worker-1"])
NFCLIENT = nf.make_client()

Example using NorFab with context manager invocation:

from norfab.core.nfapi import NorFab

with NorFab(inventory=inventory) as nf:
   ret = nf.client.run_job("nornir", "get_version")

Source code in norfab\core\nfapi.py

def __init__(
    self,
    inventory: str = "./inventory.yaml",
    inventory_data: dict = None,
    base_dir: str = None,
    log_level: str = None,
    run_broker: bool = True,
    run_workers: Union[bool, list[str]] = True,
    load_env_override: bool = True,
    configure_logging: bool = False,
    logging_name: str = "nfapi",
) -> None:
    self.log_level = log_level
    self.inventory_path = inventory
    self.inventory_data = inventory_data
    self.base_dir = base_dir
    self.exiting = False  # flag to signal that Norfab is exiting
    self.load_env_file(inventory, inventory_data, base_dir, load_env_override)
    if configure_logging:
        self.setup_logging(name=logging_name)

    self.inventory = NorFabInventory(
        path=inventory, data=inventory_data, base_dir=base_dir
    )
    if configure_logging:
        self.setup_logging(name=logging_name)

    self.run_broker = run_broker
    self.run_workers = run_workers
    self.broker_endpoint = self.inventory.broker["endpoint"]
    self.workers_init_timeout = self.inventory.topology.get(
        "workers_init_timeout", 300
    )
    self.broker_exit_event = Event()
    self.workers_exit_event = Event()
    self.clients_exit_event = Event()

    # create needed folders to kickstart the logs
    os.makedirs(
        os.path.join(self.inventory.base_dir, "__norfab__", "files"), exist_ok=True
    )
    os.makedirs(
        os.path.join(self.inventory.base_dir, "__norfab__", "logs"), exist_ok=True
    )

    # to fix ValueError: signal only works in main thread of the main interpreter
    # when trying to use nfapi to instantiate a client from different process
    try:
        signal.signal(signal.SIGINT, self.handle_ctrl_c)
    except Exception:
        pass

    # find all workers plugins
    self.register_plugins()

load_env_file(inventory: str, inventory_data: dict, base_dir: str, load_env_override: bool) -> None

Detect and load an env file before inventory initialization.

Source code in norfab\core\nfapi.py

def load_env_file(
    self,
    inventory: str,
    inventory_data: dict,
    base_dir: str,
    load_env_override: bool,
) -> None:
    """Detect and load an env file before inventory initialization."""
    env_base_dir = (
        os.path.abspath(base_dir or os.getcwd())
        if inventory_data
        else os.path.dirname(os.path.abspath(inventory))
    )
    env_file = os.path.join(env_base_dir, ".env")
    load_dotenv(dotenv_path=env_file, override=load_env_override)

resolve_base_dir(inventory: str = None, inventory_data: dict = None, base_dir: str = None) -> str staticmethod

Resolve the NorFab base directory without constructing inventory.

Source code in norfab\core\nfapi.py

@staticmethod
def resolve_base_dir(
    inventory: str = None, inventory_data: dict = None, base_dir: str = None
) -> str:
    """Resolve the NorFab base directory without constructing inventory."""
    if inventory_data:
        return os.path.abspath(base_dir or os.getcwd())

    path = os.path.abspath(inventory or "./inventory.yaml")
    return base_dir or os.path.split(path)[0]

list_environment_variables() -> dict staticmethod

Return a copy of the environment variables visible to NFAPI.

Source code in norfab\core\nfapi.py

@staticmethod
def list_environment_variables() -> dict:
    """Return a copy of the environment variables visible to NFAPI."""
    return dict(os.environ)

register_plugins() -> None

Registers worker plugins by iterating through the entry points in the 'norfab.workers' group and registering each worker plugin.

This method loads each entry point and registers it using the register_worker_plugin method.

Raises:

Type	Description
Exception	Any exceptions raised by the entry point loading or registration process.

Source code in norfab\core\nfapi.py

def register_plugins(self) -> None:
    """
    Registers worker plugins by iterating through the entry points in the
    'norfab.workers' group and registering each worker plugin.

    This method loads each entry point and registers it using the
    `register_worker_plugin` method.

    Raises:
        Exception: Any exceptions raised by the entry point loading or registration process.
    """
    # register worker plugins from entrypoints
    eps = entry_points()
    for entry_point in eps.select(group="norfab.workers"):
        self.register_worker_plugin(entry_point.name, entry_point)

    # register worker plugins from inventory
    for service_name, service_data in self.inventory.plugins.items():
        if service_data.get("worker"):
            self.register_worker_plugin(service_name, service_data["worker"])

register_worker_plugin(service_name: str, worker_plugin: Union[EntryPoint, str]) -> None

Registers a worker plugin for a given service.

This method registers a worker plugin under the specified service name. If a plugin is already registered under the same service name and it is different from the provided plugin, an exception is raised.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the service to register the plugin for.	required
worker_plugin	object	The worker plugin to be registered.	required

Raises:

Type	Description
Exception	If a different plugin is already registered under the same service name.

Source code in norfab\core\nfapi.py

def register_worker_plugin(
    self, service_name: str, worker_plugin: Union[EntryPoint, str]
) -> None:
    """
    Registers a worker plugin for a given service.

    This method registers a worker plugin under the specified service name.
    If a plugin is already registered under the same service name and it is
    different from the provided plugin, an exception is raised.

    Args:
        service_name (str): The name of the service to register the plugin for.
        worker_plugin (object): The worker plugin to be registered.

    Raises:
        Exception: If a different plugin is already registered under the same service name.
    """
    existing_plugin = self.worker_plugins.get(service_name)
    if existing_plugin is None:
        self.worker_plugins[service_name] = worker_plugin
    else:
        log.debug(
            f"Worker plugin {worker_plugin} can't be registered for "
            f"service '{service_name}' because plugin '{existing_plugin}' "
            f"was already registered under this service."
        )

handle_ctrl_c(signum, frame) -> None

Handle the CTRL-C signal (SIGINT) to gracefully exit the application.

This method is called when the user interrupts the program with a CTRL-C signal. It logs the interruption, performs necessary cleanup by calling self.destroy(), and then signals termination to the main process.

Parameters:

Name	Type	Description	Default
signum	int	The signal number (should be SIGINT).	required
frame	FrameType	The current stack frame.	required

Note

This method reassigns the SIGINT signal to the default handler and sends the SIGINT signal to the current process to ensure proper termination.

Source code in norfab\core\nfapi.py

def handle_ctrl_c(self, signum, frame) -> None:
    """
    Handle the CTRL-C signal (SIGINT) to gracefully exit the application.

    This method is called when the user interrupts the program with a CTRL-C
    signal. It logs the interruption, performs necessary cleanup by calling
    `self.destroy()`, and then signals termination to the main process.

    Args:
        signum (int): The signal number (should be SIGINT).
        frame (FrameType): The current stack frame.

    Note:
        This method reassigns the SIGINT signal to the default handler and
        sends the SIGINT signal to the current process to ensure proper
        termination.
    """
    if self.exiting is False:
        msg = "CTRL-C, NorFab exiting, interrupted by user..."
        print(f"\n{msg}")
        log.info(msg)
        self.destroy()
        # signal termination to main process
        signal.signal(signal.SIGINT, signal.default_int_handler)
        os.kill(os.getpid(), signal.SIGINT)

setup_logging(name: str = 'nfapi') -> dict

Explicitly configure logging for the current process.

If inventory is not initialized yet, bootstrap logging with NorFab defaults anchored to the resolved base directory. Once inventory exists, reapply logging with the inventory logging configuration.

Parameters:

Name	Type	Description	Default
name	str	Process identity to use in the NorFab JSONL log filename.	'nfapi'

Returns:

Name	Type	Description
dict	dict	Logging configuration applied to this process.

Source code in norfab\core\nfapi.py

def setup_logging(self, name: str = "nfapi") -> dict:
    """
    Explicitly configure logging for the current process.

    If inventory is not initialized yet, bootstrap logging with NorFab
    defaults anchored to the resolved base directory. Once inventory exists,
    reapply logging with the inventory logging configuration.

    Args:
        name: Process identity to use in the NorFab JSONL log filename.

    Returns:
        dict: Logging configuration applied to this process.
    """
    inventory = getattr(self, "inventory", None)
    if inventory is None:
        base_dir = self.resolve_base_dir(
            inventory=getattr(self, "inventory_path", None),
            inventory_data=getattr(self, "inventory_data", None),
            base_dir=getattr(self, "base_dir", None),
        )
        inventory_logging = None
    else:
        base_dir = inventory.base_dir
        inventory_logging = inventory.logging

    return setup_process_logging(
        base_dir=base_dir,
        role="nfapi",
        name=name,
        log_level=self.log_level,
        inventory_logging=inventory_logging,
    )

start_broker() -> None

Starts the broker process if a broker endpoint is defined. This method initializes and starts a separate process for the broker using the provided broker endpoint. It waits for the broker to signal that it has fully initiated, with a timeout of 30 seconds. If the broker fails to start within this time, the method logs an error message and raises a SystemExit exception.

Raises:

Type	Description
SystemExit	If the broker fails to start within 30 seconds.

Logs

Info: When the broker starts successfully. Error: If no broker endpoint is defined or if the broker fails to start.

Source code in norfab\core\nfapi.py

def start_broker(self) -> None:
    """
    Starts the broker process if a broker endpoint is defined.
    This method initializes and starts a separate process for the broker using the
    provided broker endpoint. It waits for the broker to signal that it has fully
    initiated, with a timeout of 30 seconds. If the broker fails to start within
    this time, the method logs an error message and raises a SystemExit exception.

    Raises:
        SystemExit: If the broker fails to start within 30 seconds.

    Logs:
        Info: When the broker starts successfully.
        Error: If no broker endpoint is defined or if the broker fails to start.
    """
    if self.broker_endpoint:
        init_done_event = Event()  # for worker to signal if its fully initiated

        self.broker = Process(
            target=start_broker_process,
            args=(
                self.broker_endpoint,
                self.broker_exit_event,
                self.inventory,
                self.log_level,
                init_done_event,
            ),
        )
        self.broker.start()

        # wait for broker to start
        start_time = time.time()
        while 30 > time.time() - start_time:
            if init_done_event.is_set():
                break
            time.sleep(0.1)
        else:
            log.info(
                f"Broker failed to start in 30 seconds on '{self.broker_endpoint}'"
            )
            raise SystemExit()

        log.info(
            f"Started broker, broker listening for connections on '{self.broker_endpoint}'"
        )
    else:
        log.error("Failed to start broker, no broker endpoint defined")

start_worker(worker_name, worker_data) -> None

Starts a worker process if it is not already running.

Parameters:

Name	Type	Description	Default
worker_name	str	The name of the worker to start.	required
worker_data	dict	A dictionary containing data about the worker, including any dependencies.	required

Raises:

Type	Description
RuntimeError	If a dependent process is not alive.
ServicePluginNotRegistered	If no worker plugin is registered for the worker's service.

Returns:

Type	Description
None	None

Source code in norfab\core\nfapi.py

def start_worker(self, worker_name, worker_data) -> None:
    """
    Starts a worker process if it is not already running.

    Args:
        worker_name (str): The name of the worker to start.
        worker_data (dict): A dictionary containing data about the worker, including any dependencies.

    Raises:
        RuntimeError: If a dependent process is not alive.
        norfab_exceptions.ServicePluginNotRegistered: If no worker plugin is registered for the worker's service.

    Returns:
        None
    """
    if not self.workers_processes.get(worker_name):
        log.debug(f"NFAPI PID {os.getpid()} {worker_name} starting worker process")
        worker_inventory = self.inventory[worker_name]
        init_done_event = Event()  # for worker to signal if its fully initiated

        # check dependent processes
        if worker_data.get("depends_on"):
            # check if all dependent processes are alive
            for w in worker_data["depends_on"]:
                if self.workers_processes.get(w):
                    if not self.workers_processes[w]["process"].is_alive():
                        raise RuntimeError(f"Dependent process is dead '{w}'")
            # check if all depended process fully initialized
            if not all(
                (
                    self.workers_processes[w]["init_done"].is_set()
                    if w in self.workers_processes
                    else False
                )
                for w in worker_data["depends_on"]
            ):
                return

        if not self.worker_plugins.get(worker_inventory["service"]):
            raise norfab_exceptions.ServicePluginNotRegistered(
                f"No worker plugin registered for service '{worker_inventory['service']}'"
            )
        worker_plugin = self.worker_plugins[worker_inventory["service"]]

        self.workers_processes[worker_name] = {
            "process": Process(
                target=start_worker_process,
                args=(
                    worker_plugin,
                    self.inventory,
                    self.broker_endpoint,
                    worker_name,
                    self.workers_exit_event,
                    self.log_level,
                    init_done_event,
                ),
            ),
            "init_done": init_done_event,
        }

        self.workers_processes[worker_name]["process"].start()

        log.debug(f"NFAPI PID {os.getpid()} {worker_name} worker process started")

start(run_broker: bool = None, run_workers: Union[bool, list] = None) -> None

Starts the broker and specified workers.

Parameters:

Name	Type	Description	Default
run_broker	bool	If True, starts the broker if it is defined in the inventory topology.	None
run_workers	Union[bool, list]	Determines which workers to start. If True, starts all workers defined in the inventory topology. If False or None, no workers are started. If a list, starts the specified workers.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
KeyError	If a worker fails to start due to missing inventory data.
FileNotFoundError	If a worker fails to start because the inventory file is not found.
Exception	If a worker fails to start due to any other error.

Notes

• The method waits for all workers to initialize within a specified timeout period.
• If the initialization timeout expires, an error is logged and the system is destroyed.
• After starting the workers, any startup hooks defined in the inventory are executed.

Source code in norfab\core\nfapi.py

def start(
    self,
    run_broker: bool = None,
    run_workers: Union[bool, list] = None,
) -> None:
    """
    Starts the broker and specified workers.

    Args:
        run_broker (bool): If True, starts the broker if it is defined in the inventory topology.
        run_workers (Union[bool, list]): Determines which workers to start. If True, starts all workers defined in the inventory topology.
                                     If False or None, no workers are started. If a list, starts the specified workers.

    Returns:
        None

    Raises:
        KeyError: If a worker fails to start due to missing inventory data.
        FileNotFoundError: If a worker fails to start because the inventory file is not found.
        Exception: If a worker fails to start due to any other error.

    Notes:
        - The method waits for all workers to initialize within a specified timeout period.
        - If the initialization timeout expires, an error is logged and the system is destroyed.
        - After starting the workers, any startup hooks defined in the inventory are executed.
    """
    run_broker = run_broker or self.run_broker
    run_workers = run_workers or self.run_workers
    workers_to_start = set()

    # start the broker
    if run_broker is True and self.inventory.topology.get("broker") is True:
        # update inventory to include build in services
        self.add_built_in_workers_inventory()
        self.start_broker()

    # decide on a set of workers to start
    if run_workers is False or run_workers is None:
        run_workers = []
    elif isinstance(run_workers, list) and run_workers:
        run_workers = [w.strip() for w in run_workers if w.strip()]
    # start workers defined in inventory
    elif run_workers is True:
        run_workers = self.inventory.topology.get("workers", [])

    # exit if no workers
    if not run_workers:
        return

    # form a list of workers to start
    for worker_name in run_workers:
        if isinstance(worker_name, dict):
            worker_name = tuple(worker_name)[0]
        if worker_name:
            workers_to_start.add(worker_name)
        else:
            log.error(f"'{worker_name}' - worker name is bad, skipping..")
            continue

    while workers_to_start != set(self.workers_processes.keys()):
        for worker in run_workers:
            # extract worker name and data/params
            if isinstance(worker, dict):
                worker_name = tuple(worker)[0]
                worker_data = worker[worker_name]
            elif worker:
                worker_name = worker
                worker_data = {}
            else:
                continue
            # verify if need to start this worker
            if worker_name not in workers_to_start:
                continue
            # start worker
            try:
                self.start_worker(worker_name, worker_data)
            # if failed to start remove from workers to start
            except KeyError:
                workers_to_start.discard(worker_name)
                log.error(
                    f"'{worker_name}' - failed to start worker, no inventory data found"
                )
            except FileNotFoundError as e:
                workers_to_start.discard(worker_name)
                log.error(
                    f"'{worker_name}' - failed to start worker, inventory file not found '{e}'"
                )
            except Exception as e:
                workers_to_start.discard(worker_name)
                log.exception(
                    f"'{worker_name}' - failed to start worker, error '{e}'"
                )

        time.sleep(0.01)

    # wait for workers to initialize
    start_time = time.time()
    while self.workers_init_timeout > time.time() - start_time:
        if all(w["init_done"].is_set() for w in self.workers_processes.values()):
            break
    else:
        log.error(
            f"TimeoutError - {self.workers_init_timeout}s workers init timeout expired"
        )
        self.destroy()

    # run startup hooks
    for f in self.inventory.hooks.get("startup", []):
        f["function"](self, *f.get("args", []), **f.get("kwargs", {}))

run() -> None

Runs the main loop until a termination signal (CTRL+C) is received. This method checks if there are any broker or worker processes running. If none are detected, it logs a critical message and exits. Otherwise, it enters a loop that continues to run until the exiting flag is set to True.

Source code in norfab\core\nfapi.py

def run(self) -> None:
    """
    Runs the main loop until a termination signal (CTRL+C) is received.
    This method checks if there are any broker or worker processes running.
    If none are detected, it logs a critical message and exits.
    Otherwise, it enters a loop that continues to run until the `exiting` flag is set to True.
    """
    if not self.broker and not self.workers_processes:
        log.critical(
            "NorFab detected no broker or worker processes running, exiting.."
        )
        return

    while self.exiting is False:
        time.sleep(0.1)

destroy() -> None

Gracefully stop all NORFAB processes and clean up resources.

This method performs the following steps:

1. Executes any registered exit hooks.
2. Sets the exiting flag to indicate that NORFAB is shutting down.
3. Stops all client processes.
4. Stops all worker processes and waits for them to terminate.
5. Stops the broker process and waits for it to terminate.
6. Stops the logging queue listener.

Returns:

Type	Description
None	None

Source code in norfab\core\nfapi.py

def destroy(self) -> None:
    """
    Gracefully stop all NORFAB processes and clean up resources.

    This method performs the following steps:

    1. Executes any registered exit hooks.
    2. Sets the `exiting` flag to indicate that NORFAB is shutting down.
    3. Stops all client processes.
    4. Stops all worker processes and waits for them to terminate.
    5. Stops the broker process and waits for it to terminate.
    6. Stops the logging queue listener.

    Returns:
        None
    """
    # run exit hooks
    for f in self.inventory.hooks.get("exit", []):
        f["function"](self, *f.get("args", []), **f.get("kwargs", {}))

    if self.exiting is not True:
        self.exiting = True  # indicate that NorFab already exiting
        # stop client
        log.info("NorFab is exiting, stopping clients")
        self.clients_exit_event.set()
        if self.client:
            self.client.destroy()
        # stop workers
        log.info("NorFab is exiting, stopping workers")
        self.workers_exit_event.set()
        while self.workers_processes:
            wname, w = self.workers_processes.popitem()
            w["process"].join()
            log.info(f"NorFab is exiting, stopped {wname} worker")
        # stop broker
        log.info("NorFab is exiting, stopping broker")
        self.broker_exit_event.set()
        if self.broker:
            self.broker.join()
        log.info("NorFab is exiting")

make_client(broker_endpoint: str = None, name: str = None) -> NFPClient

Creates and returns an NFPClient instance.

Parameters:

Name	Type	Description	Default
broker_endpoint	str	The broker endpoint to connect to. If not provided, the instance's broker_endpoint attribute will be used.	None
name	str	Client name to use as the broker identity and local job database folder. If not provided, defaults to '__NFPClient'.	None

Returns:

Name	Type	Description
NFPClient	NFPClient	The created client instance if a broker endpoint is defined.
None	NFPClient	If no broker endpoint is defined.

Notes

If this is the first client being created, it will be assigned to the instance's client attribute.

Source code in norfab\core\nfapi.py

def make_client(self, broker_endpoint: str = None, name: str = None) -> NFPClient:
    """
    Creates and returns an NFPClient instance.

    Args:
        broker_endpoint (str, optional): The broker endpoint to connect to.
            If not provided, the instance's broker_endpoint attribute will be used.
        name (str, optional): Client name to use as the broker identity and
            local job database folder. If not provided, defaults to
            '<entry script filename>_<current folder hash>_NFPClient'.

    Returns:
        NFPClient: The created client instance if a broker endpoint is defined.
        None: If no broker endpoint is defined.

    Notes:
        If this is the first client being created, it will be assigned to the
        instance's client attribute.
    """
    if broker_endpoint or self.broker_endpoint:
        if name is None:
            filename = os.path.splitext(os.path.basename(sys.argv[0]))[0]
            cwd_hash = hashlib.sha256(
                os.path.abspath(os.getcwd()).encode("utf-8")
            ).hexdigest()[:6]
            name = (
                f"{filename}_{cwd_hash}_NFPClient"
                if filename
                else f"{cwd_hash}_NFPClient"
            )
        client = NFPClient(
            self.inventory,
            broker_endpoint or self.broker_endpoint,
            name,
            self.clients_exit_event,
        )
        if self.client is None:  # own the first client
            self.client = client
        return client
    else:
        log.error("Failed to make client, no broker endpoint defined")
        return None