Skip to content

Best Practices

Guidelines for writing maintainable, reliable function blocks.

Naming and Packaging

Use descriptive names that communicate what the block does, not how it does it.

Aspect Convention Example
Block name Lowercase, underscore-separated verb-noun config_backup, interface_compliance_check
Package fb.{team}.neops.io fb.netops.neops.io
Version Semantic versioning tuple (1, 0, 0) for initial release

Increment the major version when you change the parameter or result schema in a breaking way. Increment minor for new optional parameters. Increment patch for bug fixes.

File Structure

Keep one function block per file (or a closely related pair, like a check and its remediation). A typical file contains:

  1. Parameter model (FunctionBlockParams subclass)
  2. Result model (FunctionBlockResultData subclass)
  3. Registration decorator
  4. Function block class

Keep the run() method focused. If it grows beyond ~40 lines, extract helper methods. Use @run_in_thread for any extracted method that performs blocking I/O.

Logging

The framework provides self.logger on every function block instance. It is typed BaseLogger and is backed by loguru unless you pass your own logger.

Structured context

Use child() to derive a logger for one part of your block. Its lines carry the name you give it as class, plus any key-value pairs you pass:

log = self.logger.child("ConfigBackup", device=device.hostname, ip=device.ip)
log.info("Starting configuration backup")
log.warning("Timeout on first attempt, retrying")

Job context

While a job runs, every SDK log line carries job_id, workflow_execution_id, step_label, function_block_id, job_type, worker_id and the entity the job runs on (entity_type, entity_id). Lines from a connection proxy, plugin or connection name their device instead: entity_type='device' and the device’s entity_id. That covers self.logger, the connection proxies and plugins, and module-level Logger() instances. You do not need to pass logger=self.logger for that; pass it only to hand on context you added yourself. The context follows the job into @run_in_thread calls, but not into a threading.Thread you start yourself.

child() binds fields to one logger. log_context adds them for a block of code, to every logger that emits inside it, including the proxies and plugins you call. To add fields for one section of work, wrap it in log_context:

from neops_worker_sdk.logger import log_context

with log_context(site=params.site):
    await self.backup_all(devices)  # every line inside carries site=...

Context values are printed on every line and pushed to the engine with it. The value of a key that names a credential (password, secret, token, passphrase, api_key, and the deployment’s NEOPS_LOG_REDACT_EXTRA_KEYS) is masked, in the context and in key=value or key: value pairs of a message. A credential under any other name is not, so never put credentials into log_context or child.

A context key reaches the engine only if it is a lowercase name of at most 64 characters ([a-z][a-z0-9_]*) and not one of the keys the log record reserves (job_id, step_label, entity_id, application, task, …); a line carries at most 32 such keys, each value cut to 2048 bytes. Any other key stays on stdout, and the worker warns once per key.

Log levels

Level When to use
debug Detailed diagnostic data (parameter values, intermediate results)
info Normal operation milestones (connected, collected N lines, completed)
warning Recoverable issues (retry succeeded, fallback used, deprecated parameter)
error Failures that cause success=False (connection refused, validation failed)

Avoid logging at error level and then returning success=True — this creates confusing signals for operators.

Parameter Design

Defaults and optionality

Make parameters optional with sensible defaults whenever the common case does not require user input:

class ComplianceCheckParams(FunctionBlockParams):
    severity_threshold: str = "warning"
    include_interfaces: bool = True
    custom_rules: list[str] | None = None

Validation

Use Pydantic validators to catch bad input early:

from pydantic import field_validator

class DeployParams(FunctionBlockParams):
    target_version: str

    @field_validator("target_version")
    @classmethod
    def validate_version(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("target_version must not be empty")
        return v.strip()

Extra fields and forward compatibility

FunctionBlockParams uses extra="ignore" in its Pydantic model config. If the workflow engine sends fields that your parameter model does not define, they are silently dropped. The SDK’s internal parse_parameters_from_dto method (which is @final) logs a warning listing the unexpected fields before Pydantic processes them. If you call MyParams.model_validate(dto) directly in tests, this SDK-level warning is bypassed — Pydantic silently drops the extras without logging. This enables forward compatibility — older function block versions continue to work when the engine adds new fields. ACQUIRE jobs take a separate, lenient path (parse_acquire_parameters_from_dto) for the reasons above; only EXECUTE uses the strict parse described here — see Parameters are best-effort at acquire time.

Common Pitfalls

Blocking I/O without @run_in_thread

Calling netmiko, paramiko, or requests directly inside run() blocks the job’s event loop. The blocking detector will warn you, but the fix is straightforward:

# Wrong — blocks the event loop
async def run(self, params, context):
    conn = ConnectHandler(host=device.ip, ...)
    output = conn.send_command("show version")

# Correct — offloaded to a thread
@run_in_thread
def _get_version(self, ip: str) -> str:
    conn = ConnectHandler(host=ip, ...)
    return conn.send_command("show version")

async def run(self, params, context):
    output = await self._get_version(device.ip)

Unchecked None on context entities

The WorkflowContext convenience properties (context.device, context.device_group, context.interface) return None when the entity is not available. Always guard access:

device = context.device
if device is None:
    return FunctionBlockResult(
        message="No device in context.", success=False, data=None
    )

Hardcoded credentials

Device credentials are available on the context entity. Never hardcode them:

# Wrong
device_params = {"username": "admin", "password": "admin123"}

# Correct
device_params = {
    "username": context.device.username,
    "password": context.device.password,
}

Credentials in the context come from the CMS and can be rotated centrally without modifying function block code.

Overly broad exception handling

Catching Exception silently hides bugs. If you need a catch-all, log the full exception:

try:
    config = await self._fetch_config(device.ip)
except Exception:
    self.logger.exception("Failed to fetch config")
    return FunctionBlockResult(
        message="Configuration fetch failed.", success=False, data=None
    )