Worker Base

Job(worker: object = None, juuid: str = None, client_address: str = None, timeout: int = None, args: list = None, kwargs: dict = None, task: str = None, client_input_queue: object = None)

Source code in norfab\core\worker.py

def __init__(
    self,
    worker: object = None,
    juuid: str = None,
    client_address: str = None,
    timeout: int = None,
    args: list = None,
    kwargs: dict = None,
    task: str = None,
    client_input_queue: object = None,
) -> None:
    self.worker = worker
    self.juuid = juuid
    self.client_address = client_address
    self.timeout = timeout
    self.args = args or []
    self.kwargs = kwargs or {}
    self.task = task
    self.client_input_queue = client_input_queue
    self.start_time = time.time()
    self.pending_input_request = None

is_timed_out() -> bool

Check if the job has exceeded its timeout.

Returns:

Name	Type	Description
bool	bool	True if the job has timed out, False otherwise or if timeout is not set.

Source code in norfab\core\worker.py

def is_timed_out(self) -> bool:
    """Check if the job has exceeded its timeout.

    Returns:
        bool: True if the job has timed out, False otherwise or if timeout is not set.
    """
    if self.timeout is None:
        return False
    return (time.time() - self.start_time) >= self.timeout

check_timeout() -> None

Raise TimeoutError if the job has exceeded its configured timeout.

Jobs without a timeout do not expire. Running operations are not interrupted.

Raises:

Type	Description
TimeoutError	The elapsed job runtime has reached its timeout.

Source code in norfab\core\worker.py

def check_timeout(self) -> None:
    """Raise TimeoutError if the job has exceeded its configured timeout.

    Jobs without a timeout do not expire. Running operations are not interrupted.

    Raises:
        TimeoutError: The elapsed job runtime has reached its timeout.
    """
    if self.is_timed_out():
        raise TimeoutError(
            f"job '{self.juuid}' task '{self.task}' exceeded its "
            f"{self.timeout}-second timeout"
        )

event(message: str, **kwargs: Any) -> None

Handles an event by forwarding it to the worker.

Parameters:

Name	Type	Description	Default
message	str	The message describing the event.	required
**kwargs	Any	Additional keyword arguments to include in the event.	{}

Source code in norfab\core\worker.py

def event(self, message: str, **kwargs: Any) -> None:
    """
    Handles an event by forwarding it to the worker.

    Args:
        message (str): The message describing the event.
        **kwargs: Additional keyword arguments to include in the event.
    """
    kwargs.setdefault("task", self.task)
    kwargs.setdefault("event_type", "progress")
    if self.juuid and self.worker:
        self.worker.event(
            message=message,
            juuid=self.juuid,
            client_address=self.client_address,
            **kwargs,
        )

stream(data: bytes) -> None

Streams data to the broker.

This method sends a message containing the client address, a unique identifier (UUID), a status code, and the provided data to the broker.

Parameters:

Name	Type	Description	Default
data	bytes	The data to be streamed to the broker.	required

Source code in norfab\core\worker.py

def stream(self, data: bytes) -> None:
    """
    Streams data to the broker.

    This method sends a message containing the client address, a unique
    identifier (UUID), a status code, and the provided data to the broker.

    Args:
        data (bytes): The data to be streamed to the broker.
    """
    msg = [
        self.client_address.encode("utf-8"),
        b"",
        self.juuid.encode("utf-8"),
        b"200",
        data,
    ]
    self.worker.send_to_broker(NFP.STREAM, msg)

wait_client_input(timeout: int = 10) -> Any

Waits for input from the client within a specified timeout period if no item is available within the specified timeout, it returns None.

Parameters:

Name	Type	Description	Default
timeout	int	The maximum time (in seconds) to wait for input	10

Returns:

Name	Type	Description
Any	Any	The item retrieved from the client_input_queue

Source code in norfab\core\worker.py

def wait_client_input(self, timeout: int = 10) -> Any:
    """
    Waits for input from the client within a specified timeout period if no item
    is available within the specified timeout, it returns `None`.

    Args:
        timeout (int, optional): The maximum time (in seconds) to wait for input

    Returns:
        Any: The item retrieved from the `client_input_queue`
    """
    try:
        return self.client_input_queue.get(block=True, timeout=timeout)
    except queue.Empty:
        pass

    return None

Task(input: Optional[BaseModel] = None, output: Optional[BaseModel] = None, description: Optional[str] = None, fastapi: Optional[dict] = None, mcp: Optional[dict] = None, agent: Optional[dict] = None)

Validate is a class-based decorator that accept arguments, designed to validate the input arguments of a task function using a specified Pydantic model. It ensures that the arguments passed to the decorated function conform to the schema defined in the model.

Attributes:

Name	Type	Description
model	BaseModel	A Pydantic model used to validate the function arguments.
name	str	The name of the task, which is used to register the task for calling, by default set equal to the name of decorated function.
result_model	BaseModel	A Pydantic model used to validate the function's return value.
fastapi	dict	Dictionary with parameters for FastAPI app.add_api_route method
mcp	dict	Dictionary with parameters for MCP mcp.types.Tool class. The optional prompts key is a NorFab extension containing a list of MCP prompt definitions associated with the task. Task validates and normalizes these definitions before registration. FastMCP removes this key before constructing the MCP Tool object. The optional guardrails key is a NorFab extension containing a list of MCP task call guardrails. FastMCP validates these definitions during discovery, removes this key before constructing the MCP Tool object, and evaluates guardrails before dispatching tool calls to NorFab services. The optional result_guardrails key is a NorFab extension containing a list of model-facing task result size and content guardrails. FastMCP validates these definitions during discovery, removes this key before constructing the MCP Tool object, and evaluates them on a copied result after task execution.

Methods:

Name	Description
__call__	Callable) -> Callable: Wraps the target function and validates its arguments before execution.
merge_args_to_kwargs	List, kwargs: Dict) -> Dict: Merges positional arguments (args) and keyword arguments (kwargs) into a single dictionary, mapping positional arguments to their corresponding parameter names based on the function's signature.
validate_input	List, kwargs: Dict) -> None: Validates merged arguments against Pydantic model. If validation fails, an exception is raised.

Usage

@Task()(input=YourPydanticModel) def your_function(arg1, arg2, ...): # Function implementation pass

Notes

• The decorator uses inspect.getfullargspec to analyze the function's signature and properly map arguments for validation.

Source code in norfab\core\worker.py

def __init__(
    self,
    input: Optional[BaseModel] = None,
    output: Optional[BaseModel] = None,
    description: Optional[str] = None,
    fastapi: Optional[dict] = None,
    mcp: Optional[dict] = None,
    agent: Optional[dict] = None,
) -> None:
    self.input = input
    self.output = output or Result
    self.description = description
    if fastapi is False:
        self.fastapi = False
    else:
        self.fastapi = fastapi or {}
    if mcp is False:
        self.mcp = False
    else:
        self.mcp = dict(mcp or {})
        if "prompts" in self.mcp:
            validated_prompts = TaskMCPPrompts(prompts=self.mcp["prompts"])
            self.mcp["prompts"] = validated_prompts.model_dump()["prompts"]
    if agent is False:
        self.agent = False
    else:
        self.agent = agent or {}

__call__(function: Callable) -> Callable

Decorator to register a function as a worker task with input/output validation and optional argument filtering.

This method wraps the provided function, validates its input arguments and output, and registers it as a task. It also removes 'job' and 'progress' keyword arguments if the wrapped function does not accept them.

Side Effects:

- Sets self.function, self.description, and self.name based on the provided function.
- Initializes input model if not already set.
- Updates the global NORFAB_WORKER_TASKS with the task schema.

Source code in norfab\core\worker.py

def __call__(self, function: Callable) -> Callable:
    """
    Decorator to register a function as a worker task with input/output
    validation and optional argument filtering.

    This method wraps the provided function, validates its input arguments
    and output, and registers it as a task. It also removes 'job' and
    'progress' keyword arguments if the wrapped function does not accept them.

    Side Effects:

        - Sets self.function, self.description, and self.name based on the provided function.
        - Initializes input model if not already set.
        - Updates the global NORFAB_WORKER_TASKS with the task schema.
    """
    self.function = function
    self.description = self.description or function.__doc__
    self.name = function.__name__

    if self.input is None:
        self.make_input_model()

    @functools.wraps(self.function)
    def wrapper(*args: object, **kwargs: object):
        # remove `job` argument if function does not expect it
        if self.is_need_argument(function, "job") is False:
            _ = kwargs.pop("job", None)

        # remove `progress` argument if function does not expect it
        if self.is_need_argument(function, "progress") is False:
            _ = kwargs.pop("progress", None)

        # validate input arguments
        self.validate_input(args, kwargs)

        ret = self.function(*args, **kwargs)

        # validate result
        self.validate_output(ret)

        return ret

    NORFAB_WORKER_TASKS.update(self.make_task_schema(wrapper))

    log.debug(
        f"{function.__module__} PID {os.getpid()} registered task '{function.__name__}'"
    )

    return wrapper

make_input_model() -> None

Dynamically creates a Pydantic input model for the worker's function by inspecting its signature.

This method uses inspect.getfullargspec to extract the function's argument names, default values, keyword-only arguments, and type annotations. It then constructs a dictionary of field specifications, giving preference to type annotations where available, and excluding special parameters such as 'self', 'return', 'job', and any args or *kwargs. The resulting specification is used to create a Pydantic model, which is assigned to self.input.

The generated model used for input validation.

Source code in norfab\core\worker.py

def make_input_model(self) -> None:
    """
    Dynamically creates a Pydantic input model for the worker's function by inspecting its signature.

    This method uses `inspect.getfullargspec` to extract the function's argument names, default values,
    keyword-only arguments, and type annotations. It then constructs a dictionary of field specifications,
    giving preference to type annotations where available, and excluding special parameters such as 'self',
    'return', 'job', and any *args or **kwargs. The resulting specification is used to create a Pydantic
    model, which is assigned to `self.input`.

    The generated model used for input validation.
    """
    (
        fun_args,  # list of the positional parameter names
        fun_varargs,  # name of the * parameter or None
        fun_varkw,  # name of the ** parameter or None
        fun_defaults,  # tuple of default argument values of the last n positional parameters
        fun_kwonlyargs,  # list of keyword-only parameter names
        fun_kwonlydefaults,  # dictionary mapping kwonlyargs parameter names to default values
        fun_annotations,  # dictionary mapping parameter names to annotations
    ) = inspect.getfullargspec(self.function)

    # form a dictionary keyed by args with their default values
    args_with_defaults = dict(
        zip(reversed(fun_args or []), reversed(fun_defaults or []))
    )

    # form a dictionary keyed by args that has no defaults with values set to
    # (Any, None) tuple if make_optional is True else set to (Any, ...)
    args_no_defaults = {
        k: (Any, ...) for k in fun_args if k not in args_with_defaults
    }

    # form dictionary keyed by args with annotations and tuple values
    args_with_hints = {
        k: (v, args_with_defaults.get(k, ...)) for k, v in fun_annotations.items()
    }

    # form merged kwargs giving preference to type hint annotations
    merged_kwargs = {**args_no_defaults, **args_with_defaults, **args_with_hints}

    # form final dictionary of fields
    fields_spec = {
        k: v
        for k, v in merged_kwargs.items()
        if k not in ["self", "return", "job", fun_varargs, fun_varkw]
    }

    log.debug(
        f"NorFab worker {self.name} task creating Pydantic input "
        f"model using fields spec: {fields_spec}"
    )

    # create Pydantic model
    self.input = create_model(self.name, **fields_spec)

make_task_schema(wrapper) -> dict

Generates a task schema dictionary for the current worker.

Parameters:

Name	Type	Description	Default
wrapper	Callable	The function wrapper to be associated with the task.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing the task's metadata, including: - function: The provided wrapper function. - module: The module name where the original function is defined. - schema: A dictionary with the following keys: - name (str): The name of the task. - description (str): The description of the task. - inputSchema (dict): The JSON schema for the input model. - outputSchema (dict): The JSON schema for the output model. - fastapi: FastAPI-specific metadata. - mcp: Model Context protocol metadata

Source code in norfab\core\worker.py

def make_task_schema(self, wrapper) -> dict:
    """
    Generates a task schema dictionary for the current worker.

    Args:
        wrapper (Callable): The function wrapper to be associated with the task.

    Returns:
        dict: A dictionary containing the task's metadata, including:
            - function: The provided wrapper function.
            - module: The module name where the original function is defined.
            - schema: A dictionary with the following keys:
                - name (str): The name of the task.
                - description (str): The description of the task.
                - inputSchema (dict): The JSON schema for the input model.
                - outputSchema (dict): The JSON schema for the output model.
                - fastapi: FastAPI-specific metadata.
                - mcp: Model Context protocol metadata
    """
    input_json_schema = self.input.model_json_schema(by_alias=False)
    _ = input_json_schema.pop("title")
    output_json_schema = self.output.model_json_schema(by_alias=False)

    return {
        self.name: {
            "function": wrapper,
            "module": self.function.__module__,
            "schema": {
                "name": str(self.name),
                "description": self.description,
                "inputSchema": input_json_schema,
                "outputSchema": output_json_schema,
                "fastapi": self.fastapi,
                "mcp": self.mcp,
                "agent": self.agent,
            },
        }
    }

is_need_argument(function: callable, argument: str) -> bool

Determines whether a given argument name is required by the function.

Source code in norfab\core\worker.py

def is_need_argument(self, function: callable, argument: str) -> bool:
    """
    Determines whether a given argument name is required by the function.
    """
    fun_args, *_ = inspect.getfullargspec(function)
    return argument in fun_args

merge_args_to_kwargs(args: List, kwargs: Dict) -> Dict

Merges positional arguments (args) and keyword arguments (kwargs) into a single dictionary.

This function uses the argument specification of the decorated function to ensure that all arguments are properly combined into a dictionary. This is particularly useful for scenarios where **kwargs need to be passed to another function or model (e.g., for validation purposes).

Parameters:

Name	Type	Description	Default
args	list	A list of positional arguments passed to the decorated function.	required
kwargs	dict	A dictionary of keyword arguments passed to the decorated function.	required

Return

dict: A dictionary containing the merged arguments, where positional arguments are mapped to their corresponding parameter names.

Source code in norfab\core\worker.py

def merge_args_to_kwargs(self, args: List, kwargs: Dict) -> Dict:
    """
    Merges positional arguments (`args`) and keyword arguments (`kwargs`)
    into a single dictionary.

    This function uses the argument specification of the decorated function
    to ensure that all arguments are properly combined into a dictionary.
    This is particularly useful for scenarios where **kwargs need to be passed
    to another function or model (e.g., for validation purposes).

    Arguments:
        args (list): A list of positional arguments passed to the decorated function.
        kwargs (dict): A dictionary of keyword arguments passed to the decorated function.

    Return:
        dict: A dictionary containing the merged arguments, where positional arguments
              are mapped to their corresponding parameter names.
    """
    merged_kwargs = {}

    (
        fun_args,  # list of the positional parameter names
        fun_varargs,  # name of the * parameter or None
        *_,  # ignore the rest
    ) = inspect.getfullargspec(self.function)

    # "def foo(a, b):" - combine "foo(1, 2)" args with "a, b" fun_args
    args_to_kwargs = dict(zip(fun_args, args))

    # "def foo(a, *b):" - combine "foo(1, 2, 3)" 2|3 args with "*b" fun_varargs
    if fun_varargs:
        args_to_kwargs[fun_varargs] = args[len(fun_args) :]

    merged_kwargs = {**kwargs, **args_to_kwargs}

    # remove reference to self if decorating class method
    _ = merged_kwargs.pop("self", None)

    return merged_kwargs

validate_input(args: List, kwargs: Dict) -> None

Function to validate provided arguments against model

Source code in norfab\core\worker.py

def validate_input(self, args: List, kwargs: Dict) -> None:
    """Function to validate provided arguments against model"""
    merged_kwargs = self.merge_args_to_kwargs(args, kwargs)
    log.debug(f"{self.name} validating input arguments: {merged_kwargs}")
    # if below step succeeds, kwargs passed model validation
    _ = self.input(**merged_kwargs)
    log.debug(
        f"Validated input kwargs: {merged_kwargs} for function {self.function} using model {self.input}"
    )

JobDatabase(db_path: str, jobs_compress: bool = True)

Thread-safe SQLite database manager for worker jobs.

Handles all job persistence operations with proper thread safety through connection-level locking and WAL mode for concurrent reads.

Attributes:

Name	Type	Description
db_path	str	Path to the SQLite database file.
_local	local	Thread-local storage for database connections.
_lock	Lock	Lock for write operations to ensure thread safety.

Initialize the job database.

Parameters:

Name	Type	Description	Default
db_path	str	Path to the SQLite database file.	required
jobs_compress	bool	If True, compress args, kwargs, and result_data fields. Defaults to True.	True

Source code in norfab\core\worker.py

def __init__(self, db_path: str, jobs_compress: bool = True) -> None:
    """
    Initialize the job database.

    Args:
        db_path (str): Path to the SQLite database file.
        jobs_compress (bool): If True, compress args, kwargs, and result_data fields. Defaults to True.
    """
    self.db_path = db_path
    self.jobs_compress = jobs_compress
    self._local = threading.local()
    self._lock = threading.Lock()
    self._initialize_database()

add_job(uuid: str, client_address: str, task: str, args: list, kwargs: dict, timeout: int, timestamp: str) -> None

Add a new job to the database.

Parameters:

Name	Type	Description	Default
uuid	str	Job UUID.	required
client_address	str	Client address.	required
task	str	Task name.	required
args	list	Task arguments.	required
kwargs	dict	Task keyword arguments.	required
timeout	int	Job timeout.	required
timestamp	str	Received timestamp.	required

Source code in norfab\core\worker.py

def add_job(
    self,
    uuid: str,
    client_address: str,
    task: str,
    args: list,
    kwargs: dict,
    timeout: int,
    timestamp: str,
) -> None:
    """
    Add a new job to the database.

    Args:
        uuid (str): Job UUID.
        client_address (str): Client address.
        task (str): Task name.
        args (list): Task arguments.
        kwargs (dict): Task keyword arguments.
        timeout (int): Job timeout.
        timestamp (str): Received timestamp.
    """
    now = time.strftime("%Y-%m-%d %H:%M:%S")
    with self._transaction(write=True) as conn:
        # Serialize args and kwargs to BLOB
        compressed_args = self._compress_data({"args": args})
        compressed_kwargs = self._compress_data({"kwargs": kwargs})

        conn.execute(
            """
            INSERT INTO jobs (uuid, client_address, task, args, kwargs, timeout,
                             status, received_timestamp, created_at)
            VALUES (?, ?, ?, ?, ?, ?, 'PENDING', ?, ?)
        """,
            (
                uuid,
                client_address,
                task,
                compressed_args,
                compressed_kwargs,
                timeout,
                timestamp,
                now,
            ),
        )

get_next_pending_job() -> tuple

Get the next pending job and mark it as STARTED.

Returns:

Name	Type	Description
tuple	tuple	(uuid, received_timestamp) or None if no pending jobs.

Source code in norfab\core\worker.py

def get_next_pending_job(self) -> tuple:
    """
    Get the next pending job and mark it as STARTED.

    Returns:
        tuple: (uuid, received_timestamp) or None if no pending jobs.
    """
    with self._transaction(write=True) as conn:
        # order jobs by their creation timestamp in ascending order - oldest first
        cursor = conn.execute("""
            SELECT uuid, received_timestamp FROM jobs
            WHERE status = 'PENDING'
            ORDER BY created_at ASC
            LIMIT 1
        """)
        row = cursor.fetchone()
        if row:
            uuid = row["uuid"]
            conn.execute(
                """
                UPDATE jobs SET status = 'STARTED', started_timestamp = ?
                WHERE uuid = ?
            """,
                (time.ctime(), uuid),
            )
            return uuid, row["received_timestamp"]
        return None

complete_job(uuid: str, result_data: dict) -> None

Mark a job as completed and store its result.

Parameters:

Name	Type	Description	Default
uuid	str	Job UUID.	required
result_data	dict	Result data as dictionary.	required

Source code in norfab\core\worker.py

def complete_job(self, uuid: str, result_data: dict) -> None:
    """
    Mark a job as completed and store its result.

    Args:
        uuid (str): Job UUID.
        result_data (dict): Result data as dictionary.
    """
    try:
        compressed_result_data = self._compress_data(result_data)
    except Exception as e:
        for wname, wres in result_data["result"].items():
            wres["errors"] = [
                f"Worker experienced error while compressing job results data: '{e}'"
            ]
            wres["failed"] = True
            wres["result"] = {}
        self.fail_job(uuid, result_data)
    else:
        with self._transaction(write=True) as conn:
            conn.execute(
                """
                UPDATE jobs
                SET status = 'COMPLETED', completed_timestamp = ?, result_data = ?
                WHERE uuid = ?
            """,
                (time.ctime(), compressed_result_data, uuid),
            )

fail_job(uuid: str, result_data: dict) -> None

Mark a job as failed and store its result.

Parameters:

Name	Type	Description	Default
uuid	str	Job UUID.	required
result_data	dict	Result data as dictionary.	required

Source code in norfab\core\worker.py

def fail_job(self, uuid: str, result_data: dict) -> None:
    """
    Mark a job as failed and store its result.

    Args:
        uuid (str): Job UUID.
        result_data (dict): Result data as dictionary.
    """
    with self._transaction(write=True) as conn:
        conn.execute(
            """
            UPDATE jobs
            SET status = 'FAILED', completed_timestamp = ?, result_data = ?
            WHERE uuid = ?
            """,
            (time.ctime(), self._compress_data(result_data), uuid),
        )

get_job_info(uuid: str, include_result: bool = False, include_events: bool = False) -> dict

Get comprehensive job information including status, execution data, and optionally result data and events.

Parameters:

Name	Type	Description	Default
uuid	str	Job UUID.	required
include_result	bool	If True, include result_data in the response. Defaults to False.	False
include_events	bool	If True, include job events. Defaults to False.	False

Returns:

Name	Type	Description
dict	dict	Job information with the following fields: - uuid: Job UUID - status: Job status (PENDING, STARTED, COMPLETED, FAILED, WAITING_CLIENT_INPUT) - received_timestamp: When job was received - started_timestamp: When job started execution - completed_timestamp: When job completed - client_address: Client address - task: Task name - args: Parsed task arguments list - kwargs: Parsed task keyword arguments dict - timeout: Job timeout
	dict	If include_result=True, also includes: - result_data: Result data dictionary (if available)
	dict	If include_events=True, also includes: - job_events: List of event dictionaries
	dict	Returns None if job not found.

Source code in norfab\core\worker.py

def get_job_info(
    self,
    uuid: str,
    include_result: bool = False,
    include_events: bool = False,
) -> dict:
    """
    Get comprehensive job information including status, execution data, and optionally result data and events.

    Args:
        uuid (str): Job UUID.
        include_result (bool): If True, include result_data in the response. Defaults to False.
        include_events (bool): If True, include job events. Defaults to False.

    Returns:
        dict: Job information with the following fields:
            - uuid: Job UUID
            - status: Job status (PENDING, STARTED, COMPLETED, FAILED, WAITING_CLIENT_INPUT)
            - received_timestamp: When job was received
            - started_timestamp: When job started execution
            - completed_timestamp: When job completed
            - client_address: Client address
            - task: Task name
            - args: Parsed task arguments list
            - kwargs: Parsed task keyword arguments dict
            - timeout: Job timeout

        If include_result=True, also includes:
            - result_data: Result data dictionary (if available)

        If include_events=True, also includes:
            - job_events: List of event dictionaries

        Returns None if job not found.
    """
    with self._transaction(write=False) as conn:
        # Build SELECT clause based on requested fields
        select_fields = [
            "uuid",
            "status",
            "received_timestamp",
            "started_timestamp",
            "completed_timestamp",
            "client_address",
            "task",
            "args",
            "kwargs",
            "timeout",
        ]

        if include_result:
            select_fields.append("result_data")

        query = f"""
            SELECT {', '.join(select_fields)}
            FROM jobs WHERE uuid = ?
        """

        cursor = conn.execute(query, (uuid,))
        row = cursor.fetchone()

        if not row:
            return None

        result = dict(row)
        result.setdefault("result_data", None)
        result.setdefault("job_events", [])

        # Deserialize args and kwargs from BLOB
        result["args"] = self._decompress_data(row["args"]).get("args", [])
        result["kwargs"] = self._decompress_data(row["kwargs"]).get("kwargs", {})

        # Include result data if requested
        if include_result and row["result_data"]:
            result["result_data"] = self._decompress_data(row["result_data"])

        # Include events if requested
        if include_events:
            result["job_events"] = self.get_job_events(uuid)

        return result

add_event(job_uuid: str, message: str, severity: str, task: str, event_data: dict) -> None

Add an event for a job.

Parameters:

Name	Type	Description	Default
job_uuid	str	Job UUID.	required
message	str	Event message.	required
severity	str	Event severity.	required
task	str	Task name.	required
event_data	dict	Event data dictionary.	required

Source code in norfab\core\worker.py

def add_event(
    self, job_uuid: str, message: str, severity: str, task: str, event_data: dict
) -> None:
    """
    Add an event for a job.

    Args:
        job_uuid (str): Job UUID.
        message (str): Event message.
        severity (str): Event severity.
        task (str): Task name.
        event_data (dict): Event data dictionary.
    """
    with self._transaction(write=True) as conn:
        conn.execute(
            """
            INSERT INTO events (job_uuid, message, severity, task, event_data)
            VALUES (?, ?, ?, ?, ?)
        """,
            (
                job_uuid,
                message,
                severity,
                task,
                orjson.dumps(event_data).decode("utf-8"),
            ),
        )

get_job_events(uuid: str) -> list

Get all events for a job.

Parameters:

Name	Type	Description	Default
uuid	str	Job UUID.	required

Returns:

Name	Type	Description
list	list	List of event dictionaries.

Source code in norfab\core\worker.py

def get_job_events(self, uuid: str) -> list:
    """
    Get all events for a job.

    Args:
        uuid (str): Job UUID.

    Returns:
        list: List of event dictionaries.
    """
    with self._transaction(write=False) as conn:
        cursor = conn.execute(
            """
            SELECT message, severity, task, event_data, created_at
            FROM events WHERE job_uuid = ?
            ORDER BY created_at ASC
        """,
            (uuid,),
        )
        return [
            {
                "message": row["message"],
                "severity": row["severity"],
                "task": row["task"],
                **orjson.loads(row["event_data"]),
                "created_at": row["created_at"],
            }
            for row in cursor.fetchall()
        ]

list_jobs(pending: bool = True, completed: bool = True, task: str = None, last: int = None, client: str = None, uuid: str = None) -> list

List jobs based on filters.

Parameters:

Name	Type	Description	Default
pending	bool	Include pending jobs.	True
completed	bool	Include completed jobs.	True
task	str	Filter by task name.	None
last	int	Return only last N jobs.	None
client	str	Filter by client address.	None
uuid	str	Filter by specific UUID.	None

Returns:

Name	Type	Description
list	list	List of job dictionaries.

Source code in norfab\core\worker.py

def list_jobs(
    self,
    pending: bool = True,
    completed: bool = True,
    task: str = None,
    last: int = None,
    client: str = None,
    uuid: str = None,
) -> list:
    """
    List jobs based on filters.

    Args:
        pending (bool): Include pending jobs.
        completed (bool): Include completed jobs.
        task (str): Filter by task name.
        last (int): Return only last N jobs.
        client (str): Filter by client address.
        uuid (str): Filter by specific UUID.

    Returns:
        list: List of job dictionaries.
    """
    with self._transaction(write=False) as conn:
        # Build query
        conditions = []
        params = []

        if uuid:
            conditions.append("uuid = ?")
            params.append(uuid)
        else:
            status_conditions = []
            if pending:
                status_conditions.append(
                    "status IN ('PENDING', 'STARTED', 'WAITING_CLIENT_INPUT')"
                )
            if completed:
                status_conditions.append("status IN ('COMPLETED', 'FAILED')")
            if status_conditions:
                conditions.append(f"({' OR '.join(status_conditions)})")

            if task:
                conditions.append("task = ?")
                params.append(task)
            if client:
                conditions.append("client_address = ?")
                params.append(client)

        where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
        query = f"""
            SELECT uuid, client_address, task, status,
                   received_timestamp, started_timestamp, completed_timestamp
            FROM jobs {where_clause}
            ORDER BY created_at DESC
        """

        if last:
            query += f" LIMIT {last}"

        cursor = conn.execute(query, params)
        return [dict(row) for row in cursor.fetchall()]

jobs_stats() -> dict

Return counts of all retained jobs, grouped by status.

Counts persist across restarts and decrease when job history is removed. Failed jobs are counted separately from completed jobs.

Returns:

Name	Type	Description
dict	dict	Total job count and a mapping of statuses to job counts.

Source code in norfab\core\worker.py

def jobs_stats(self) -> dict:
    """Return counts of all retained jobs, grouped by status.

    Counts persist across restarts and decrease when job history is removed.
    Failed jobs are counted separately from completed jobs.

    Returns:
        dict: Total job count and a mapping of statuses to job counts.
    """
    with self._transaction(write=False) as conn:
        rows = conn.execute("SELECT status, COUNT(*) FROM jobs GROUP BY status")
        jobs_by_status = {row[0]: row[1] for row in rows.fetchall()}
    return {
        "total_jobs": sum(jobs_by_status.values()),
        "jobs_by_status": jobs_by_status,
    }

close() -> None

Close all database connections.

Source code in norfab\core\worker.py

def close(self) -> None:
    """Close all database connections."""
    if hasattr(self._local, "conn"):
        self._local.conn.close()
        delattr(self._local, "conn")

NFPWorker(inventory: NorFabInventory, broker: str, service: str, name: str, exit_event: object, log_level: str = None, multiplier: int = 6, keepalive: int = 2500)

NFPWorker class is responsible for managing worker operations, including connecting to a broker, handling jobs, and maintaining keepalive connections. It interacts with the broker using ZeroMQ and manages job queues and events.

Parameters:

Name	Type	Description	Default
inventory	NorFabInventory	The inventory object containing base directory information.	required
broker	str	The broker address.	required
service	str	The service name.	required
name	str	The name of the worker.	required
exit_event	object	The event used to signal the worker to exit.	required
log_level	str	Logging level configured from worker inventory.	None
multiplier	int	The multiplier value. Defaults to 6.	6
keepalive	int	The keepalive interval in milliseconds. Defaults to 2500.	2500

Source code in norfab\core\worker.py

def __init__(
    self,
    inventory: NorFabInventory,
    broker: str,
    service: str,
    name: str,
    exit_event: object,
    log_level: str = None,
    multiplier: int = 6,
    keepalive: int = 2500,
) -> None:
    self.inventory = inventory
    self.name = name
    self.setup_logging(log_level)
    self.max_concurrent_jobs = max(1, inventory.get("max_concurrent_jobs", 5))
    self.jobs_compress = inventory.get("jobs_compress", True)
    self.autostart_watchdog = inventory.get("autostart_watchdog", True)
    self.broker = broker
    self.service = service.encode("utf-8") if isinstance(service, str) else service
    self.exit_event = exit_event
    self.broker_socket = None
    self.multiplier = multiplier
    self.keepalive = keepalive
    self.zmq_auth = self.inventory.broker.get("zmq_auth", True)
    self.build_message = NFP.MessageBuilder()
    self.monitoring = ProcessMonitor()
    self.public_keys_dir = None
    self.secret_keys_dir = None
    self.worker_private_key_file = None
    self.broker_public_key_file = None

    # create base directories
    self.base_dir = os.path.join(
        self.inventory.base_dir, "__norfab__", "files", "worker", self.name
    )
    os.makedirs(self.base_dir, exist_ok=True)

    # Initialize SQLite database for job management
    db_path = os.path.join(self.base_dir, f"{self.name}.db")
    self.db = JobDatabase(db_path, jobs_compress=self.jobs_compress)

    # dictionary to store currently running jobs
    self.running_jobs = {}
    self.status = {
        "sid_inventory_status": None,
    }

    # create events and queues
    self.destroy_event = threading.Event()
    self.pending_job_event = threading.Event()
    self.request_thread = None
    self.reply_thread = None
    self.recv_thread = None
    self.event_thread = None
    self.put_thread = None

    self.post_queue = queue.Queue(maxsize=0)
    self.get_queue = queue.Queue(maxsize=0)
    self.delete_queue = queue.Queue(maxsize=0)
    self.event_queue = queue.Queue(maxsize=0)
    self.put_queue = queue.Queue(maxsize=0)
    self.outbound_queue = queue.Queue(maxsize=NFP.OUTBOUND_QUEUE_SIZE)

    # generate certificates and create directories
    if self.zmq_auth is not False:
        generate_certificates(
            self.base_dir,
            cert_name=self.name,
            broker_keys_dir=os.path.join(
                self.inventory.base_dir,
                "__norfab__",
                "files",
                "broker",
                "public_keys",
            ),
            inventory=self.inventory,
        )
        self.public_keys_dir = os.path.join(self.base_dir, "public_keys")
        self.secret_keys_dir = os.path.join(self.base_dir, "private_keys")

    self.ctx = zmq.Context()
    self.poller = zmq.Poller()
    self.reconnect_to_broker()

    self.client = NFPClient(
        self.inventory,
        self.broker,
        name=f"{self.name}-NFPClient",
        exit_event=self.exit_event,
    )

    self.tasks = NORFAB_WORKER_TASKS

    # initiate watchdog
    if self.autostart_watchdog and self.autostart_base_watchdog:
        self.watchdog = WorkerWatchDog(self)
        self.watchdog.start()

setup_logging(log_level: str = None) -> dict

Configure logging for this worker process.

The worker uses the logging inventory it received during construction as a per-process template and writes its default NorFab file sink to __norfab__/logs/worker-<worker-name>.jsonl.

Parameters:

Name	Type	Description	Default
log_level	str	Optional logging level override.	None

Returns:

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

Source code in norfab\core\worker.py

def setup_logging(self, log_level: str = None) -> dict:
    """
    Configure logging for this worker process.

    The worker uses the logging inventory it received during construction
    as a per-process template and writes its default NorFab file sink to
    ``__norfab__/logs/worker-<worker-name>.jsonl``.

    Args:
        log_level: Optional logging level override.

    Returns:
        dict: Logging configuration applied to this process.
    """
    return setup_process_logging(
        base_dir=self.inventory.base_dir,
        role="worker",
        name=self.name,
        log_level=log_level,
        inventory_logging=self.inventory.logging,
    )

reconnect_to_broker(queue_ready: bool = True) -> list | None

Connect or reconnect to the broker.

This method handles the connection or reconnection process to the broker. It performs the following steps:

1. If there is an existing broker socket, unregisters it from the poller and closes it.
2. Creates a new DEALER socket and sets its identity.
3. Loads the client's secret and public keys for CURVE authentication.
4. Loads the server's public key for CURVE authentication.
5. Connects the socket to the broker.
6. Registers the socket with the poller for incoming messages.
7. Builds a READY message to register the service.
8. Starts or restarts the keepalive mechanism to maintain the connection.
9. Increments the reconnect statistics counter.
10. Logs the successful registration to the broker.

Parameters:

Name	Type	Description	Default
queue_ready	bool	If True, queue the READY frame. If False, return the READY frame so the socket owner can send it immediately before any queued worker messages.	True

Source code in norfab\core\worker.py

def reconnect_to_broker(self, queue_ready: bool = True) -> list | None:
    """
    Connect or reconnect to the broker.

    This method handles the connection or reconnection process to the broker.
    It performs the following steps:

    1. If there is an existing broker socket, unregisters it from the poller
       and closes it.
    2. Creates a new DEALER socket and sets its identity.
    3. Loads the client's secret and public keys for CURVE authentication.
    4. Loads the server's public key for CURVE authentication.
    5. Connects the socket to the broker.
    6. Registers the socket with the poller for incoming messages.
    7. Builds a READY message to register the service.
    8. Starts or restarts the keepalive mechanism to maintain the connection.
    9. Increments the reconnect statistics counter.
    10. Logs the successful registration to the broker.

    Args:
        queue_ready: If True, queue the READY frame. If False, return the
            READY frame so the socket owner can send it immediately before
            any queued worker messages.
    """
    if self.broker_socket:
        self.poller.unregister(self.broker_socket)
        self.broker_socket.close()

    self.broker_socket = self.ctx.socket(zmq.DEALER)
    self.broker_socket.setsockopt_unicode(zmq.IDENTITY, self.name, "utf8")
    self.broker_socket.linger = 0

    if self.zmq_auth is not False:
        # We need two certificates, one for the client and one for
        # the server. The client must know the server's public key
        # to make a CURVE connection.
        self.worker_private_key_file = os.path.join(
            self.secret_keys_dir, f"{self.name}.key_secret"
        )
        client_public, client_secret = zmq.auth.load_certificate(
            self.worker_private_key_file
        )
        self.broker_socket.curve_secretkey = client_secret
        self.broker_socket.curve_publickey = client_public

        # The client must know the server's public key to make a CURVE connection.
        self.broker_public_key_file = os.path.join(
            self.public_keys_dir, "broker.key"
        )
        server_public, _ = zmq.auth.load_certificate(self.broker_public_key_file)
        self.broker_socket.curve_serverkey = server_public

    self.broker_socket.connect(self.broker)
    self.poller.register(self.broker_socket, zmq.POLLIN)

    # Register service with broker
    ready_msg = self.build_message.worker_to_broker_ready(service=self.service)
    if queue_ready:
        self.outbound_queue.put(
            ready_msg, block=True, timeout=NFP.OUTBOUND_QUEUE_TIMEOUT
        )
    log.debug(f"{self.name} - NFP.READY prepared for broker '{self.broker}'")

    if self.keepaliver is not None:
        self.keepaliver.restart()
    else:
        self.keepaliver = KeepAliver(
            address=None,
            multiplier=self.multiplier,
            keepalive=self.keepalive,
            exit_event=self.destroy_event,
            service=self.service,
            whoami=NFP.WORKER,
            name=self.name,
        )
        self.keepaliver.start()

    self.monitoring.record_reconnect()
    log.info(
        f"{self.name} - registered to broker at '{self.broker}', "
        f"service '{self.service.decode('utf-8')}'"
    )
    return None if queue_ready else ready_msg

send_to_broker(command: str, msg: list = None) -> None

Send a message to the broker.

Parameters:

Name	Type	Description	Default
command	str	The command to send to the broker. Must be one of NFP.READY, NFP.DISCONNECT, NFP.RESPONSE, or NFP.EVENT.	required
msg	list	The message to send. If not provided, a default message will be created based on the command.	None

Logs

Logs an error if the command is unsupported. Logs a debug message with the message being sent.

Thread Safety

This method is thread-safe because non-owner threads only submit prepared messages to the outbound queue. The zmq_send_recv thread owns the ZeroMQ socket.

Source code in norfab\core\worker.py

def send_to_broker(self, command: str, msg: list = None) -> None:
    """
    Send a message to the broker.

    Parameters:
        command (str): The command to send to the broker. Must be one of NFP.READY, NFP.DISCONNECT, NFP.RESPONSE, or NFP.EVENT.
        msg (list, optional): The message to send. If not provided, a default message will be created based on the command.

    Logs:
        Logs an error if the command is unsupported.
        Logs a debug message with the message being sent.

    Thread Safety:
        This method is thread-safe because non-owner threads only submit
        prepared messages to the outbound queue. The zmq_send_recv thread
        owns the ZeroMQ socket.
    """
    if command == NFP.READY:
        msg = self.build_message.worker_to_broker_ready(service=self.service)
    elif command == NFP.DISCONNECT:
        msg = self.build_message.worker_to_broker_disconnect(service=self.service)
    elif command == NFP.RESPONSE:
        msg = self.build_message.worker_to_broker_response(response_data=msg)
    elif command == NFP.EVENT:
        msg = self.build_message.worker_to_broker_event(event_data=msg)
    elif command == NFP.STREAM:
        msg = self.build_message.worker_to_broker_stream(data=msg)
    else:
        log.error(
            f"{self.name} - cannot send '{command}' to broker, command unsupported"
        )
        return

    log.debug(f"{self.name} - queueing '{msg}'")

    if self.destroy_event.is_set():
        return

    try:
        self.outbound_queue.put_nowait(msg)
    except queue.Full:
        log.warning(
            f"{self.name} - outbound queue is full, dropping '{command}' "
            f"message to broker"
        )

load_inventory() -> dict

Load inventory data from the broker for this worker.

This function retrieves inventory data from the broker service using the worker's name. It logs the received inventory data and returns the results if available.

Returns:

Name	Type	Description
dict	dict	The inventory data results if available, otherwise an empty dictionary.

Source code in norfab\core\worker.py

def load_inventory(self) -> dict:
    """
    Load inventory data from the broker for this worker.

    This function retrieves inventory data from the broker service using the worker's name.
    It logs the received inventory data and returns the results if available.

    Returns:
        dict: The inventory data results if available, otherwise an empty dictionary.
    """
    self.status["sid_inventory_status"] = "initialising"
    try:
        inventory_data = self.client.mmi(
            "sid.service.broker", "get_inventory", kwargs={"name": self.name}
        )
    except Exception:
        self.status["sid_inventory_status"] = "failed"
        raise

    log.debug(f"{self.name} - worker received inventory data {inventory_data}")

    self.status["sid_inventory_status"] = (
        "failed" if inventory_data.get("errors") else "completed"
    )

    if inventory_data["results"]:
        return inventory_data["results"]
    else:
        return {}

worker_exit() -> None

Method to override in child classes with a set of actions to perform on exit call.

This method should be implemented by subclasses to define any cleanup or finalization tasks that need to be performed when the worker is exiting.

Source code in norfab\core\worker.py

def worker_exit(self) -> None:
    """
    Method to override in child classes with a set of actions to perform on exit call.

    This method should be implemented by subclasses to define any cleanup or finalization
    tasks that need to be performed when the worker is exiting.
    """
    return None

get_inventory(job: Job) -> Result

Retrieve the worker's inventory.

This method should be overridden in child classes to provide the specific implementation for retrieving the inventory of a worker.

Returns:

Name	Type	Description
Dict	Result	A dictionary representing the worker's inventory.

Raises:

Type	Description
NotImplementedError	If the method is not overridden in a child class.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]})
def get_inventory(self, job: Job) -> Result:
    """
    Retrieve the worker's inventory.

    This method should be overridden in child classes to provide the specific
    implementation for retrieving the inventory of a worker.

    Returns:
        Dict: A dictionary representing the worker's inventory.

    Raises:
        NotImplementedError: If the method is not overridden in a child class.
    """
    raise NotImplementedError

get_version() -> Result

Retrieve the version report of the worker.

This method should be overridden in child classes to provide the specific version report of the worker.

Returns:

Name	Type	Description
Dict	Result	A dictionary containing the version information of the worker.

Raises:

Type	Description
NotImplementedError	If the method is not overridden in a child class.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]})
def get_version(self) -> Result:
    """
    Retrieve the version report of the worker.

    This method should be overridden in child classes to provide the specific
    version report of the worker.

    Returns:
        Dict: A dictionary containing the version information of the worker.

    Raises:
        NotImplementedError: If the method is not overridden in a child class.
    """
    raise NotImplementedError

destroy(message: str = None) -> None

Cleanly shuts down the worker by performing the following steps:

1. Calls the worker_exit method to handle any worker-specific exit procedures.
2. Sets the destroy_event to signal that the worker is being destroyed.
3. Calls the destroy method on the client to clean up client resources.
4. Joins all the threads (request_thread, reply_thread, event_thread, recv_thread) if they are not None, ensuring they have finished execution.
5. Closes the database connections.
6. Destroys the context with a linger period of 0 to immediately close all sockets.
7. Stops the keepaliver to cease any keepalive signals.
8. Logs an informational message indicating that the worker has been destroyed, including an optional message.

Parameters:

Name	Type	Description	Default
message	str	An optional message to include in the log when the worker is destroyed.	None

Source code in norfab\core\worker.py

def destroy(self, message: str = None) -> None:
    """
    Cleanly shuts down the worker by performing the following steps:

    1. Calls the worker_exit method to handle any worker-specific exit procedures.
    2. Sets the destroy_event to signal that the worker is being destroyed.
    3. Calls the destroy method on the client to clean up client resources.
    4. Joins all the threads (request_thread, reply_thread, event_thread, recv_thread) if they are not None, ensuring they have finished execution.
    5. Closes the database connections.
    6. Destroys the context with a linger period of 0 to immediately close all sockets.
    7. Stops the keepaliver to cease any keepalive signals.
    8. Logs an informational message indicating that the worker has been destroyed, including an optional message.

    Args:
        message (str, optional): An optional message to include in the log when the worker is destroyed.
    """
    self.worker_exit()
    self.destroy_event.set()
    self.client.destroy()

    # join all the threads
    if self.request_thread is not None:
        self.request_thread.join()
    if self.reply_thread is not None:
        self.reply_thread.join()
    if self.event_thread is not None:
        self.event_thread.join()
    if self.put_thread is not None:
        self.put_thread.join()
    if self.recv_thread:
        self.recv_thread.join()

    # close database
    self.db.close()

    self.ctx.destroy(0)

    # stop keepalives
    self.keepaliver.stop()

    log.info(f"{self.name} - worker destroyed, message: '{message}'")

is_url(url: str) -> bool

Check if the given string is a URL supported by NorFab File Service.

Parameters:

Name	Type	Description	Default
url	str	The URL to check.	required

Returns:

Name	Type	Description
bool	bool	True if the URL supported by NorFab File Service, False otherwise.

Source code in norfab\core\worker.py

def is_url(self, url: str) -> bool:
    """
    Check if the given string is a URL supported by NorFab File Service.

    Args:
        url (str): The URL to check.

    Returns:
        bool: True if the URL supported by NorFab File Service, False otherwise.
    """
    return any(str(url).startswith(k) for k in ["nf://", "git://"])

fetch_file(url: str, raise_on_fail: bool = False, read: bool = True) -> str

Function to download file from broker File Sharing Service

Parameters:

Name	Type	Description	Default
url	str	file location string in nf://<filepath> or git://<remote-name>/<filepath> format	required
raise_on_fail	bool	raise FileNotFoundError if download fails	False
read	bool	if True returns file content, return OS path to saved file otherwise	True

Returns:

Name	Type	Description
str	str	File content if read is True, otherwise OS path to the saved file.

Raises:

Type	Description
FileNotFoundError	If raise_on_fail is True and the download fails.

Source code in norfab\core\worker.py

def fetch_file(
    self, url: str, raise_on_fail: bool = False, read: bool = True
) -> str:
    """
    Function to download file from broker File Sharing Service

    Args:
        url: file location string in ``nf://<filepath>`` or
            ``git://<remote-name>/<filepath>`` format
        raise_on_fail: raise FileNotFoundError if download fails
        read: if True returns file content, return OS path to saved file otherwise

    Returns:
        str: File content if read is True, otherwise OS path to the saved file.

    Raises:
        FileNotFoundError: If raise_on_fail is True and the download fails.
    """
    if not self.is_url(url):
        raise ValueError(f"Invalid URL format: {url}")

    result = self.client.fetch_file(url=url, read=read)
    status = result["status"]
    file_content = result["content"]
    msg = f"{self.name} - worker '{url}' fetch file failed with status '{status}'"

    if status == "200":
        return file_content
    elif raise_on_fail is True:
        raise FileNotFoundError(msg)
    else:
        log.error(msg)
        return None

jinja2_render_templates(templates: list[str], context: dict = None, filters: dict = None, template_cache: dict = None) -> str

Download (if needed) and render a list of Jinja2 templates with the given context and optional filters.

Parameters:

Name	Type	Description	Default
templates	list[str]	A list of Jinja2 template strings or NorFab file paths.	required
context	dict	A dictionary containing the context variables for rendering the templates.	None
filters	dict	A dictionary of custom Jinja2 filters to be used during rendering.	None
template_cache	dict	Cache of compiled Jinja2 file templates keyed by resolved NorFab file URL.	None

Returns:

Name	Type	Description
str	str	The rendered templates concatenated into a single string.

Source code in norfab\core\worker.py

def jinja2_render_templates(
    self,
    templates: list[str],
    context: dict = None,
    filters: dict = None,
    template_cache: dict = None,
) -> str:
    """
    Download (if needed) and render a list of Jinja2 templates with the given context and optional filters.

    Args:
        templates (list[str]): A list of Jinja2 template strings or NorFab file paths.
        context (dict): A dictionary containing the context variables for rendering the templates.
        filters (dict, optional): A dictionary of custom Jinja2 filters to be used during rendering.
        template_cache (dict, optional): Cache of compiled Jinja2 file templates
            keyed by resolved NorFab file URL.

    Returns:
        str: The rendered templates concatenated into a single string.
    """
    rendered = []
    filters = filters or {}
    context = context or {}
    template_cache = template_cache if template_cache is not None else {}
    for template in templates:
        if any(marker in template for marker in ("{{", "{%", "{#")):
            j2env = Environment(loader="BaseLoader")
            j2env.filters.update(filters)  # add custom filters
            renderer = j2env.from_string(template)
            template = renderer.render(**context)
        # download template file and render it again
        if self.is_url(template):
            renderer = template_cache.get(template)
            if renderer is None:
                filepath = self.jinja2_fetch_template(template)
                searchpath, filename = os.path.split(filepath)
                j2env = Environment(loader=FileSystemLoader(searchpath))
                j2env.filters.update(filters)  # add custom filters
                renderer = j2env.get_template(filename)
                template_cache[template] = renderer
            rendered.append(renderer.render(**context))
        # template content is fully rendered
        else:
            rendered.append(template)

    return "\n".join(rendered)

jinja2_fetch_template(url: str) -> str

Helper function to recursively download a Jinja2 template along with other templates referenced using "include" statements.

Parameters:

Name	Type	Description	Default
url	str	An nf:// or git:// URL to download.	required

Returns:

Name	Type	Description
str	str	The file path of the downloaded Jinja2 template.

Raises:

Type	Description
FileNotFoundError	If the file download fails.
Exception	If Jinja2 template parsing fails.

Source code in norfab\core\worker.py

def jinja2_fetch_template(self, url: str) -> str:
    """
    Helper function to recursively download a Jinja2 template along with
    other templates referenced using "include" statements.

    Args:
        url (str): An ``nf://`` or ``git://`` URL to download.

    Returns:
        str: The file path of the downloaded Jinja2 template.

    Raises:
        FileNotFoundError: If the file download fails.
        Exception: If Jinja2 template parsing fails.
    """
    filepath = self.fetch_file(url, read=False)
    if filepath is None:
        msg = f"{self.name} - file download failed '{url}'"
        raise FileNotFoundError(msg)

    # load Jinja2 template content to parse "include"-ed files
    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read()
    j2env = Environment(loader="BaseLoader")
    try:
        parsed_content = j2env.parse(content)
    except Exception as e:
        msg = f"{self.name} - Jinja2 template parsing failed '{url}', error: '{e}'"
        raise Exception(msg)

    # run recursion on include statements
    for node in parsed_content.find_all(Include):
        include_file = node.template.value
        base_url = url.rsplit("/", 1)[0]
        self.jinja2_fetch_template(f"{base_url}/{include_file}")

    return filepath

event(message: str, juuid: str, task: str, client_address: str, **kwargs: Any) -> None

Handles the creation and emission of an event.

This method takes event data, processes it, and sends it to the event queue. It also saves the event data to the database for future reference.

Parameters:

Name	Type	Description	Default
message	str	The event message	required
juuid	str	Job ID for which this event is generated	required
task	str	Task name	required
client_address	str	Client address	required
**kwargs	Any	Additional keyword arguments to be passed when creating a NorFabEvent instance	{}

Logs

Error: Logs an error message if the event data cannot be formed.

Source code in norfab\core\worker.py

def event(
    self,
    message: str,
    juuid: str,
    task: str,
    client_address: str,
    **kwargs: Any,
) -> None:
    """
    Handles the creation and emission of an event.

    This method takes event data, processes it, and sends it to the event queue.
    It also saves the event data to the database for future reference.

    Args:
        message: The event message
        juuid: Job ID for which this event is generated
        task: Task name
        client_address: Client address
        **kwargs: Additional keyword arguments to be passed when creating a NorFabEvent instance

    Logs:
        Error: Logs an error message if the event data cannot be formed.
    """
    kwargs.setdefault("event_type", "progress")
    # construct NorFabEvent
    try:
        event_data = NorFabEvent(
            message=message,
            juuid=juuid,
            client_address=client_address,
            task=task,
            **kwargs,
        )
    except Exception as e:
        log.error(f"Failed to form event data, error {e}")
        return
    event_dict = event_data.model_dump(exclude_none=True)

    # emit event to the broker
    self.event_queue.put(event_dict)

    # check if need to emit log for this event
    if self.inventory["logging"].get("log_events", False):
        event_log = f"EVENT {self.name}:{task} - {message}"
        severity = event_dict.get("severity", "INFO")
        if severity == "INFO":
            log.info(event_log)
        elif severity == "DEBUG":
            log.debug(event_log)
        elif severity == "WARNING":
            log.warning(event_log)
        elif severity == "CRITICAL":
            log.critical(event_log)
        elif severity == "ERROR":
            log.error(event_log)

    # save event to database
    try:
        self.db.add_event(
            job_uuid=juuid,
            message=message,
            severity=event_dict.get("severity", "INFO"),
            task=task,
            event_data=event_dict,
        )
    except Exception as e:
        log.error(f"Failed to save event to database: {e}")

job_details(uuid: str = None, result: bool = True, events: bool = True) -> Result

Method to get job details by UUID for completed jobs.

Parameters:

Name	Type	Description	Default
uuid	str	The job UUID to return details for.	None
result	bool	If True, return job result.	True
events	bool	If True, return job events.	True

Returns:

Name	Type	Description
Result	Result	A Result object with the job details.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def job_details(
    self,
    uuid: str = None,
    result: bool = True,
    events: bool = True,
) -> Result:
    """
    Method to get job details by UUID for completed jobs.

    Args:
        uuid (str): The job UUID to return details for.
        result (bool): If True, return job result.
        events (bool): If True, return job events.

    Returns:
        Result: A Result object with the job details.
    """
    job = self.db.get_job_info(
        uuid=uuid, include_result=result, include_events=events
    )

    if job:
        return Result(
            task=f"{self.name}:job_details",
            result=job,
        )
    else:
        raise RuntimeError(f"{self.name} - job with UUID '{uuid}' not found")

job_list(pending: bool = True, completed: bool = True, task: str = None, last: int = None, client: str = None, uuid: str = None) -> Result

Method to list worker jobs completed and pending.

Parameters:

Name	Type	Description	Default
pending	bool	If True or None, return pending jobs. If False, skip pending jobs.	True
completed	bool	If True or None, return completed jobs. If False, skip completed jobs.	True
task	str	If provided, return only jobs with this task name.	None
last	int	If provided, return only the last N completed and last N pending jobs.	None
client	str	If provided, return only jobs submitted by this client.	None
uuid	str	If provided, return only the job with this UUID.	None

Returns:

Name	Type	Description
Result	Result	Result object with a list of jobs.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def job_list(
    self,
    pending: bool = True,
    completed: bool = True,
    task: str = None,
    last: int = None,
    client: str = None,
    uuid: str = None,
) -> Result:
    """
    Method to list worker jobs completed and pending.

    Args:
        pending (bool): If True or None, return pending jobs. If False, skip pending jobs.
        completed (bool): If True or None, return completed jobs. If False, skip completed jobs.
        task (str, optional): If provided, return only jobs with this task name.
        last (int, optional): If provided, return only the last N completed and last N pending jobs.
        client (str, optional): If provided, return only jobs submitted by this client.
        uuid (str, optional): If provided, return only the job with this UUID.

    Returns:
        Result: Result object with a list of jobs.
    """
    jobs = self.db.list_jobs(
        pending=pending,
        completed=completed,
        task=task,
        last=last,
        client=client,
        uuid=uuid,
    )

    # Add worker and service information to each job
    for job in jobs:
        job["worker"] = self.name
        job["service"] = self.service.decode("utf-8")
        # Map database status to expected status names
        if job["status"] in ("PENDING", "STARTED"):
            job["status"] = "PENDING"
            job["completed_timestamp"] = None
        elif job["status"] in ("COMPLETED", "FAILED"):
            job["status"] = "COMPLETED"

    return Result(
        task=f"{self.name}:job_list",
        result=jobs,
    )

get_logs(last: int = 100, level: str = None, logger: str = None, since: str = None, until: str = None) -> Result

Return this worker process JSONL log records.

Parameters:

Name	Type	Description	Default
last	int	Return the last N records after filtering.	100
level	str	Filter by log severity.	None
logger	str	Filter by logger name.	None
since	str	Return records at or after this timestamp.	None
until	str	Return records at or before this timestamp.	None

Returns:

Name	Type	Description
Result	Result	Log records from this worker's process log file.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def get_logs(
    self,
    last: int = 100,
    level: str = None,
    logger: str = None,
    since: str = None,
    until: str = None,
) -> Result:
    """
    Return this worker process JSONL log records.

    Args:
        last: Return the last N records after filtering.
        level: Filter by log severity.
        logger: Filter by logger name.
        since: Return records at or after this timestamp.
        until: Return records at or before this timestamp.

    Returns:
        Result: Log records from this worker's process log file.
    """
    records = read_jsonl_logs(
        logs_dir=os.path.join(self.inventory.base_dir, "__norfab__", "logs"),
        log_files=[f"worker-{self.name}.jsonl"],
        last=last,
        level=level,
        logger=logger,
        since=since,
        until=until,
    )
    for record in records:
        record.setdefault("role", "worker")
        record.setdefault("name", self.name)

    return Result(
        task=f"{self.name}:get_logs",
        result=records,
    )

echo(job: Job, raise_error: Union[bool, int, str] = None, sleep: int = None, *args: Any, **kwargs: Any) -> Result

Echoes the job information and optional arguments, optionally sleeping or raising an error.

Parameters:

Name	Type	Description	Default
job	Job	The job instance containing job details.	required
raise_error	str	If provided, raises a RuntimeError with this message.	None
sleep	int	If provided, sleeps for the specified number of seconds.	None
*args	Any	Additional positional arguments to include in the result.	()
**kwargs	Any	Additional keyword arguments to include in the result.	{}

Returns:

Name	Type	Description
Result	Result	An object containing job details and any provided arguments.

Raises:

Type	Description
RuntimeError	If raise_error is provided.

Source code in norfab\core\worker.py

@Task(
    fastapi={"methods": ["POST"]},
    input=models.WorkerEchoIn,
    output=models.WorkerEchoOut,
)
def echo(
    self,
    job: Job,
    raise_error: Union[bool, int, str] = None,
    sleep: int = None,
    *args: Any,
    **kwargs: Any,
) -> Result:
    """
    Echoes the job information and optional arguments, optionally sleeping or raising an error.

    Args:
        job (Job): The job instance containing job details.
        raise_error (str, optional): If provided, raises a RuntimeError with this message.
        sleep (int, optional): If provided, sleeps for the specified number of seconds.
        *args: Additional positional arguments to include in the result.
        **kwargs: Additional keyword arguments to include in the result.

    Returns:
        Result: An object containing job details and any provided arguments.

    Raises:
        RuntimeError: If `raise_error` is provided.
    """
    if sleep:
        time.sleep(sleep)
    if raise_error:
        raise RuntimeError(raise_error)
    return Result(
        result={
            "juuid": job.juuid,
            "client_address": job.client_address,
            "timeout": job.timeout,
            "task": job.task,
            "args": args,
            "kwargs": kwargs,
        }
    )

run_shell_cmd(job: Job, command: str, timeout: int = None) -> Result

Runs a shell command on the worker host.

Parameters:

Name	Type	Description	Default
command	str	Shell command to execute.	required
timeout	int	Maximum command runtime in seconds.	None

Returns:

Name	Type	Description
Result	Result	Command execution details including stdout, stderr, and return code.

Source code in norfab\core\worker.py

@Task(
    fastapi={"methods": ["POST"]},
    mcp=False,
    agent={"enabled": False},
)
def run_shell_cmd(self, job: Job, command: str, timeout: int = None) -> Result:
    """
    Runs a shell command on the worker host.

    Args:
        command (str): Shell command to execute.
        timeout (int, optional): Maximum command runtime in seconds.

    Returns:
        Result: Command execution details including stdout, stderr, and return code.
    """
    try:
        proc = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as exc:
        return Result(
            failed=True,
            status="failed",
            errors=[f"Command timed out after {timeout} seconds"],
            result={
                "command": command,
                "stdout": exc.stdout or "",
                "stderr": exc.stderr or "",
                "returncode": None,
                "timeout": timeout,
                "timed_out": True,
            },
        )

    failed = proc.returncode != 0
    return Result(
        failed=failed,
        errors=[proc.stderr] if failed and proc.stderr else [],
        result={
            "command": command,
            "stdout": proc.stdout,
            "stderr": proc.stderr,
            "returncode": proc.returncode,
            "timeout": timeout,
            "timed_out": False,
        },
    )

list_tasks(name: Union[None, str] = None, brief: bool = False) -> Result

Lists tasks supported by worker.

Parameters:

Name	Type	Description	Default
name	str	The name of a specific task to retrieve	None
brief	bool	If True, returns only the list of task names	False

Returns:

Type	Description
Result	Results returned controlled by this logic: If brief is True returns a list of task names If name is provided returns list with single item - OpenAPI schema of the specified task Otherwise returns a list of schemas for all tasks

Raises:

Type	Description
KeyError	If a specific task name is provided but not registered in NORFAB_WORKER_TASKS.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def list_tasks(self, name: Union[None, str] = None, brief: bool = False) -> Result:
    """
    Lists tasks supported by worker.

    Args:
        name (str, optional): The name of a specific task to retrieve
        brief (bool, optional): If True, returns only the list of task names

    Returns:
        Results returned controlled by this logic:

            - If brief is True returns a list of task names
            - If name is provided returns list with single item - OpenAPI schema of the specified task
            - Otherwise returns a list of schemas for all tasks

    Raises:
        KeyError: If a specific task name is provided but not registered in NORFAB_WORKER_TASKS.
    """
    ret = Result()
    if brief:
        ret.result = list(sorted(NORFAB_WORKER_TASKS.keys()))
    elif name:
        if name not in NORFAB_WORKER_TASKS:
            raise KeyError(f"{name} - task not registered")
        ret.result = [NORFAB_WORKER_TASKS[name]["schema"]]
    else:
        ret.result = [t["schema"] for t in NORFAB_WORKER_TASKS.values()]
    return ret

get_worker_status() -> Result

Return worker identity, runtime paths, and security configuration.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def get_worker_status(self) -> Result:
    """Return worker identity, runtime paths, and security configuration."""
    return Result(
        result={
            "name": self.name,
            "role": "worker",
            "service": self.service.decode("utf-8"),
            "status": ("active" if not self.exit_event.is_set() else "stopping"),
            "broker": self.broker,
            "directories": {
                "base_dir": self.base_dir,
                "public_keys_dir": self.public_keys_dir,
                "private_keys_dir": self.secret_keys_dir,
            },
            "security": {
                "worker_private_key_file": self.worker_private_key_file,
                "broker_public_key_file": self.broker_public_key_file,
                "zmq_auth": self.zmq_auth,
            },
        }
    )

get_stats() -> Result

Return process monitoring and retained job counts.

Job counts cover the worker's SQLite history across restarts, with no time window or record limit. Removing history reduces these counts.

Returns:

Name	Type	Description
Result	Result	Validated monitoring snapshot, including total jobs and counts by status under jobs. Failed jobs are counted separately from completed jobs.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def get_stats(self) -> Result:
    """Return process monitoring and retained job counts.

    Job counts cover the worker's SQLite history across restarts, with no
    time window or record limit. Removing history reduces these counts.

    Returns:
        Result: Validated monitoring snapshot, including total jobs and
            counts by status under ``jobs``. Failed jobs are counted
            separately from completed jobs.
    """
    stats_data = {
        "name": self.name,
        "status": "active" if not self.exit_event.is_set() else "stopping",
        "process": self.monitoring.process_stats(),
        "messaging": self.monitoring.messaging_stats(),
        "service": self.service.decode("utf-8"),
        "broker": self.broker,
        "reconnects": self.monitoring.reconnects,
        "outbound_queue_depth": self.outbound_queue.qsize(),
        "running_jobs": len(self.running_jobs),
        "jobs": self.db.jobs_stats(),
        "max_concurrent_jobs": self.max_concurrent_jobs,
        "watchdog_runs": self.watchdog.runs,
        "keepalives_sent": self.keepaliver.keepalives_send,
        "keepalives_received": self.keepaliver.keepalives_received,
    }
    stats_data.update(self.watchdog.get_stats())
    stats = self.monitoring_model(
        **stats_data,
    )
    return Result(result=stats.model_dump(mode="json"))

get_watchdog_configuration() -> Result

Retrieves the current configuration of the watchdog.

Returns:

Name	Type	Description
Result	Result	An object containing the watchdog configuration.

Source code in norfab\core\worker.py

@Task(fastapi={"methods": ["GET"]}, agent={"enabled": False})
def get_watchdog_configuration(self) -> Result:
    """
    Retrieves the current configuration of the watchdog.

    Returns:
        Result: An object containing the watchdog configuration.
    """
    return Result(result=self.watchdog.configuration())

start_threads() -> None

Starts multiple daemon threads required for the worker's operation.

This method initializes and starts the following threads

• request_thread: Handles posting requests using the _post function.
• reply_thread: Handles receiving replies using the _get function.
• event_thread: Handles event processing using the _event function.
• recv_thread: Handles ZeroMQ send/receive using the zmq_send_recv function.

Each thread is started as a daemon and is provided with the necessary arguments, including queues and events as required.

Returns:

Type	Description
None	None

Source code in norfab\core\worker.py

def start_threads(self) -> None:
    """
    Starts multiple daemon threads required for the worker's operation.

    This method initializes and starts the following threads:
        - request_thread: Handles posting requests using the _post function.
        - reply_thread: Handles receiving replies using the _get function.
        - event_thread: Handles event processing using the _event function.
        - recv_thread: Handles ZeroMQ send/receive using the zmq_send_recv function.

    Each thread is started as a daemon and is provided with the necessary arguments,
    including queues and events as required.

    Returns:
        None
    """
    # Start threads
    self.request_thread = threading.Thread(
        target=_post,
        daemon=True,
        name=f"{self.name}_post_thread",
        args=(
            self,
            self.post_queue,
            self.destroy_event,
        ),
    )
    self.request_thread.start()
    self.reply_thread = threading.Thread(
        target=_get,
        daemon=True,
        name=f"{self.name}_get_thread",
        args=(self, self.get_queue, self.destroy_event),
    )
    self.reply_thread.start()
    self.event_thread = threading.Thread(
        target=_event,
        daemon=True,
        name=f"{self.name}_event_thread",
        args=(self, self.event_queue, self.destroy_event),
    )
    self.event_thread.start()
    self.put_thread = threading.Thread(
        target=_put,
        daemon=True,
        name=f"{self.name}_put_thread",
        args=(
            self,
            self.put_queue,
            self.destroy_event,
        ),
    )
    self.put_thread.start()
    # start receive thread after other threads
    self.recv_thread = threading.Thread(
        target=zmq_send_recv,
        daemon=True,
        name=f"{self.name}_zmq_send_recv_thread",
        args=(
            self,
            self.destroy_event,
        ),
    )
    self.recv_thread.start()

run_next_job(uuid: str) -> None

Processes the next job from the database.

This method performs the following steps:

1. Loads job data from the database.
2. Parses the job data to extract the task name, arguments, keyword arguments, and timeout.
3. Executes the specified task method on the worker instance with the provided arguments.
4. Handles any exceptions raised during task execution, logging errors and creating a failed Result object if needed.
5. Saves the result of the job execution to the database.
6. Marks the job as completed or failed in the database.

Parameters:

Name	Type	Description	Default
uuid	str	The job UUID to process.	required

Raises:

Type	Description
TypeError	If the executed task does not return a Result object.

Source code in norfab\core\worker.py

def run_next_job(self, uuid: str) -> None:
    """
    Processes the next job from the database.

    This method performs the following steps:

    1. Loads job data from the database.
    2. Parses the job data to extract the task name, arguments, keyword arguments, and timeout.
    3. Executes the specified task method on the worker instance with the provided arguments.
    4. Handles any exceptions raised during task execution, logging errors and creating a failed Result object if needed.
    5. Saves the result of the job execution to the database.
    6. Marks the job as completed or failed in the database.

    Args:
        uuid (str): The job UUID to process.

    Raises:
        TypeError: If the executed task does not return a Result object.
    """
    log.info(f"{self.name} - Processing job request {uuid}")

    # Load job data from database
    job_data = self.db.get_job_info(uuid)
    if not job_data:
        log.error(f"{self.name} - job {uuid} not found in database")
        return

    client_address = job_data["client_address"]
    task = job_data["task"]
    args = job_data["args"]
    kwargs = job_data["kwargs"]
    timeout = job_data["timeout"]

    job = Job(
        worker=self,
        client_address=client_address,
        juuid=uuid,
        task=task,
        timeout=timeout,
        args=copy.deepcopy(args),
        kwargs=copy.deepcopy(kwargs),
        client_input_queue=queue.Queue(maxsize=0),
    )
    self.running_jobs[uuid] = job

    log.debug(
        f"{self.name} - doing task '{task}', timeout: '{timeout}', "
        f"args: '{args}', kwargs: '{kwargs}', client: '{client_address}', "
        f"job uuid: '{uuid}'"
    )
    log.info(f"{self.name} - Starting task '{task}' for job {uuid}")

    # inform client that job started
    job.event(message="starting", status="running")

    # run the actual job
    try:
        task_started = time.ctime()
        result = NORFAB_WORKER_TASKS[task]["function"](
            self, *args, job=job, **kwargs
        )
        task_completed = time.ctime()
        if not isinstance(result, Result):
            raise TypeError(
                f"{self.name} - task '{task}' did not return Result object, "
                f"args: '{args}', kwargs: '{kwargs}', client: '{client_address}', "
                f"job uuid: '{uuid}'; task returned '{type(result)}'"
            )
        result.task = result.task or f"{self.name}:{task}"
        result.status = result.status or "completed"
        result.juuid = result.juuid or uuid
        result.service = self.service.decode("utf-8")
        job_failed = False
    except Exception as e:
        task_completed = time.ctime()
        result = Result(
            task=f"{self.name}:{task}",
            errors=[traceback.format_exc()],
            messages=[f"Worker experienced error: '{e}'"],
            failed=True,
            juuid=uuid,
        )
        log.error(
            f"{self.name} - worker experienced error:\n{traceback.format_exc()}"
        )
        job_failed = True

    result.task_started = task_started
    result.task_completed = task_completed

    # Prepare result data for database storage as JSON-serializable dict
    result_data = {
        "client_address": client_address,
        "uuid": uuid,
        "status_code": "200",
        "result": {self.name: result.model_dump()},
        "worker": self.name,
        "service": self.service.decode("utf-8"),
    }

    # Save job result to database
    if job_failed:
        self.db.fail_job(uuid, result_data)
        log.error(f"{self.name} - Task '{task}' failed for job {uuid}")
    else:
        self.db.complete_job(uuid, result_data)
        log.info(f"{self.name} - Completed task '{task}' for job {uuid}")

    # remove job from running jobs
    _ = self.running_jobs.pop(uuid)

    # inform client that job completed
    job.event(message="completed", status="completed")

work() -> None

Executes the main worker loop, managing job execution using a thread pool.

This method starts necessary background threads, then enters a loop where it:

• Queries the database for the next pending job.
• Atomically marks the job as started in the database.
• Submits the job to a thread pool executor for concurrent processing.
• Waits briefly if no pending jobs are found.
• Continues until either the exit or destroy event is set.

Upon exit, performs cleanup by calling the destroy method with a status message.

Source code in norfab\core\worker.py

def work(self) -> None:
    """
    Executes the main worker loop, managing job execution using a thread pool.

    This method starts necessary background threads, then enters a loop where it:

    - Queries the database for the next pending job.
    - Atomically marks the job as started in the database.
    - Submits the job to a thread pool executor for concurrent processing.
    - Waits briefly if no pending jobs are found.
    - Continues until either the exit or destroy event is set.

    Upon exit, performs cleanup by calling the `destroy` method with a status message.
    """

    self.start_threads()

    # start job threads and submit jobs in an infinite loop
    with concurrent.futures.ThreadPoolExecutor(
        max_workers=self.max_concurrent_jobs,
        thread_name_prefix=f"{self.name}-job-thread",
    ) as executor:
        while not self.exit_event.is_set() and not self.destroy_event.is_set():
            # Get next pending job from database
            job_info = self.db.get_next_pending_job()

            if job_info is None:
                # Sleep until a POST adds work, while retaining a bounded
                # wait so shutdown events are observed promptly.
                self.pending_job_event.wait(timeout=1)
                self.pending_job_event.clear()
                continue

            uuid, received_timestamp = job_info
            log.debug(f"{self.name} - submitting job {uuid} to executor")

            # Submit the job to workers
            executor.submit(self.run_next_job, uuid)

    # make sure to clean up
    self.destroy(
        f"{self.name} - exit event is set '{self.exit_event.is_set()}', "
        f"destroy event is set '{self.destroy_event.is_set()}'"
    )

zmq_send_recv(worker: object, destroy_event: threading.Event) -> None

ZeroMQ send/receive thread.

This function owns the worker's broker socket. Other threads submit prepared messages to the outbound queue instead of touching the socket directly. It drains outbound messages, sends due keepalives, polls the socket, and processes received messages.

Parameters:

Name	Type	Description	Default
worker	Worker	The worker instance that contains the broker socket and queues.	required
destroy_event	Event	An event to signal the thread to stop.	required

Commands

• NFP.POST: Places the message into the post_queue.
• NFP.DELETE: Places the message into the delete_queue.
• NFP.GET: Places the message into the get_queue.
• NFP.KEEPALIVE: Processes a keepalive heartbeat.
• NFP.DISCONNECT: Attempts to reconnect to the broker.
• Other: Logs an invalid input message.

Source code in norfab\core\worker.py

def zmq_send_recv(worker: object, destroy_event: threading.Event) -> None:
    """
    ZeroMQ send/receive thread.

    This function owns the worker's broker socket. Other threads submit prepared
    messages to the outbound queue instead of touching the socket directly.
    It drains outbound messages, sends due keepalives, polls the socket, and
    processes received messages.

    Args:
        worker (Worker): The worker instance that contains the broker socket and queues.
        destroy_event (threading.Event): An event to signal the thread to stop.

    Commands:
        - NFP.POST: Places the message into the post_queue.
        - NFP.DELETE: Places the message into the delete_queue.
        - NFP.GET: Places the message into the get_queue.
        - NFP.KEEPALIVE: Processes a keepalive heartbeat.
        - NFP.DISCONNECT: Attempts to reconnect to the broker.
        - Other: Logs an invalid input message.
    """
    while not destroy_event.is_set():
        for _ in range(NFP.ZMQ_SEND_RECV_DRAIN_LIMIT):
            try:
                outbound = worker.outbound_queue.get_nowait()
            except queue.Empty:
                break
            try:
                if not worker.destroy_event.is_set():
                    worker.broker_socket.send_multipart(outbound)
                    worker.monitoring.record_sent()
            except Exception as e:
                worker.monitoring.record_send_failure()
                log.error(
                    f"{worker.name} - failed to send queued message to broker, "
                    f"error '{e}'",
                    exc_info=True,
                )

        if worker.keepaliver and worker.keepaliver.due():
            try:
                if not worker.destroy_event.is_set():
                    worker.broker_socket.send_multipart(
                        worker.keepaliver.make_message()
                    )
                    worker.monitoring.record_sent()
            except Exception as e:
                worker.monitoring.record_send_failure()
                log.error(
                    f"{worker.name} - failed to send keepalive to broker, error '{e}'"
                )

        msg = None
        try:
            items = worker.poller.poll(NFP.ZMQ_SEND_RECV_POLL_TIMEOUT_MS)
            if items:
                try:
                    msg = worker.broker_socket.recv_multipart(zmq.NOBLOCK)
                except zmq.Again:
                    # Readiness can become stale; resume polling instead of blocking.
                    pass
        except KeyboardInterrupt:
            break  # Interrupted

        if msg is not None:
            worker.monitoring.record_received()
            log.debug(f"{worker.name} - received '{msg}'")
            if len(msg) < 3:
                worker.monitoring.record_receive_failure()
                log.error(f"{worker.name} - received malformed message: {msg}")
                continue
            empty = msg.pop(0)  # noqa
            header = msg.pop(0)
            command = msg.pop(0)

            if command == NFP.POST:
                worker.post_queue.put(msg)
            elif command == NFP.DELETE:
                worker.delete_queue.put(msg)
            elif command == NFP.GET:
                worker.get_queue.put(msg)
            elif command == NFP.PUT:
                worker.put_queue.put(msg)
            elif command == NFP.KEEPALIVE:
                worker.keepaliver.received_heartbeat([header] + msg)
            elif command == NFP.DISCONNECT:
                ready_msg = worker.reconnect_to_broker(queue_ready=False)
                if ready_msg and not worker.destroy_event.is_set():
                    try:
                        worker.broker_socket.send_multipart(ready_msg)
                        worker.monitoring.record_sent()
                    except Exception as e:
                        worker.monitoring.record_send_failure()
                        log.error(
                            f"{worker.name} - failed to send ready to broker, "
                            f"error '{e}'"
                        )
            else:
                log.debug(
                    f"{worker.name} - invalid input, header '{header}', command '{command}', message '{msg}'"
                )

        if not worker.keepaliver.is_alive():
            log.warning(f"{worker.name} - '{worker.broker}' broker keepalive expired")
            if not worker.destroy_event.is_set():
                disconnect_msg = worker.build_message.worker_to_broker_disconnect(
                    service=worker.service
                )
                try:
                    worker.broker_socket.send_multipart(disconnect_msg)
                    worker.monitoring.record_sent()
                except Exception as e:
                    worker.monitoring.record_send_failure()
                    log.error(
                        f"{worker.name} - failed to send disconnect to broker, "
                        f"error '{e}'"
                    )
            ready_msg = worker.reconnect_to_broker(queue_ready=False)
            if ready_msg and not worker.destroy_event.is_set():
                try:
                    worker.broker_socket.send_multipart(ready_msg)
                    worker.monitoring.record_sent()
                except Exception as e:
                    worker.monitoring.record_send_failure()
                    log.error(
                        f"{worker.name} - failed to send ready to broker, error '{e}'"
                    )