Advanced Patterns

This page covers patterns you will encounter as your function blocks grow beyond simple read-and-return operations: requesting additional data before execution, handling rollback, deciding when to split blocks, and structuring error handling.

The Acquire Phase

Every function block has an acquire() method that the workflow engine calls before run(). Its purpose is to request additional entity data that the block needs but that is not included in the default context.

sequenceDiagram
    participant WE as Workflow Engine
    participant FB as Function Block
    WE->>FB: acquire(params)
    FB-->>WE: FunctionBlockAcquireResult (acquires list)
    WE->>WE: Process acquisitions, enrich context
    WE->>FB: run(params, enriched_context)
    FB-->>WE: FunctionBlockResult

acquire() returns a FunctionBlockAcquireResult containing an optional list of EntityAcquire items. Each item is one of:

Type	Purpose
`EntityAcquireByElasticQuery`	Fetch entities matching an Elasticsearch query
`EntityAcquireContext`	Expand the context to include related entities (e.g., all interfaces of a device)

The workflow engine processes these acquisitions, merges the results into the WorkflowContext, and then calls run() with the enriched context.

If your block does not need additional data, return an empty result:

async def acquire(self, params: AcquireParams[MyParams]) -> FunctionBlockAcquireResult:
    return FunctionBlockAcquireResult(
        message="No additional data required.",
        success=True,
        acquires=None,
    )

When to use acquire

Use the acquire phase when your function block operates on a device but also needs data about its interfaces, parent group, or devices matching a dynamic query. Without acquire, the context only contains the entities the workflow engine provides by default based on run_on.

Parameters are best-effort at acquire time

acquire() receives an AcquireParams[YourParams] (from neops_worker_sdk.function_block.acquire_params), not your parameter model directly. At ACQUIRE time no earlier step has run and there is no entity context, so the engine cannot resolve parameters fed by {{ <step>.result… }} or {{ context… }}. Those fields are listed in params.unresolved; on params.partial a required one is None, while one with a declared default keeps that default (so a defaulted field is always usable, even when unresolved). Fields fed by {{ parameters… }} or written literally in the workflow are resolved and can be trusted.

async def acquire(self, params: AcquireParams[MyParams]) -> FunctionBlockAcquireResult:
    # Branch on `unresolved`, not on `is None`: for a required field the static
    # type is non-optional, so a type checker cannot see that None is possible.
    if "site" in params.unresolved:
        query = "device.id: 1"
    else:
        query = f"device.id: 1 AND site: {params.partial.site}"

    return FunctionBlockAcquireResult(
        success=True,
        message="Acquiring interfaces.",
        acquires=[
            EntityAcquireByElasticQuery(
                type="elastic", entity="interface", query=query,
                assertNotEmpty=False, description="Interfaces for the device.",
            )
        ],
    )

params.raw holds the untouched payload if you need an escape hatch.

These are not the final values

Do not treat params.partial as the parameters your run() will see. The EXECUTE phase re-interpolates every parameter from scratch and validates it against the strict schema, so a field you widen or skip here is neither lost nor permanently relaxed — but a value you read here may still differ from the one run() receives.

Rollback

The FunctionBlock base class defines a rollback() method that the workflow engine calls, through a ROLLBACK job, when the workflow fails and this block was registered reversible=True. The job carries the same parameters and entity context the forward job had; this step itself may have succeeded.

async def rollback(
    self,
    params: ParamsT,
    context: ContextT,
    forward: ForwardResult[ResultDataT, RollbackDataT],
) -> FunctionBlockRollbackResult:

The default implementation returns success=False with the message "Rollback not implemented"; the entry then ends ROLLBACK_FAILED and the run waits for a human to resolve it.

forward describes the earliest attempt of this step that may have changed something, as a whole: a retry runs on whatever the failed attempt left behind, so the earliest attempt holds the state from before the step. Attempts that reported side_effects=False are skipped; an attempt whose outcome is unknown never is.

Field	Meaning
`success`, `message`	That attempt’s result
`data`	Its result data, parsed into the block’s `result_cls`, or `None`
`rollback_data`	The rollback data it saved, parsed into the block’s `rollback_data_cls`, or `None`
`rollback_data_rejected`	Why the engine dropped its rollback data (`not_execute`, `not_reversible`, `no_schema`, `too_large`, `invalid`), or `None`
`outcome_known`	`False` when the engine wrote that attempt’s result (a timeout, a lost worker) or there is none yet; `rollback_data` is then `None`
`previous_rollbacks`	The step’s earlier rollback attempts, oldest first, each with `success`, `message` and `outcome_known`

If forward is missing from the job, or does not parse into those classes, the worker fails the rollback job with retryable=False and does not call rollback().

Declare reversibility explicitly

Only a block registered reversible=True receives rollback jobs (see Pure and idempotent blocks). Register it so only once rollback() can restore the state, from rollback data or from params and context.

Rollback undoes the device, not the CMS

The engine writes a run’s entity changes to the CMS only when the whole run succeeds, so a run that is rolled back has written none of them. There is nothing in the CMS for rollback() to restore, and changes it makes to context are not sent to the engine.

Rollback data

A reversible block can record what it needs to undo its change, such as the running configuration it is about to replace, before it changes anything. Declare the shape as a FunctionBlockRollbackData subclass, name it in the registration and pass it as the block’s last generic argument:

from neops_worker_sdk.function_block.function_block import (
    DeviceFunctionBlock,
    ForwardResult,
    FunctionBlockRollbackResult,
)
from neops_worker_sdk.function_block.rollback_data import FunctionBlockRollbackData


class Snapshot(FunctionBlockRollbackData):
    running: str


@register_function_block(
    Registration(
        ...,
        param_cls=Params,
        result_cls=Result,
        rollback_data_cls=Snapshot,
        reversible=True,
    )
)
class Configure(DeviceFunctionBlock[Params, Result, Snapshot]):
    async def run(self, params, context):
        current = await self._read_running(context.device)
        self.save_rollback_data(Snapshot(running=current))  # before any write
        ...

    async def rollback(self, params, context, forward: ForwardResult[Result, Snapshot]):
        if forward.rollback_data is None:  # this block chooses to refuse
            return FunctionBlockRollbackResult(success=False, message="No snapshot: restore by hand")
        await self._restore(context.device, forward.rollback_data.running)
        return FunctionBlockRollbackResult(success=True, message="Running configuration restored.")

The rollback data class is the fourth generic argument of FunctionBlock and the third of DeviceFunctionBlock, InterfaceFunctionBlock and GroupFunctionBlock; it defaults to NoRollbackData. The SDK checks the two declarations against each other: registration fails when the generic argument and rollback_data_cls disagree, or when rollback_data_cls is set on a block that is not reversible. The class forbids extra fields, and declaring a computed field on it is an error. Its JSON schema is registered with the block and frozen per exact version, so changing the shape means a new version. That includes giving an already registered version its first rollback_data_cls: bump the version.

save_rollback_data(data, *, replace=False):

Serializes at once. It checks the type, dumps the model to JSON, re-validates that value and keeps the JSON, not the model. Changing the model afterwards changes nothing that is sent.
Saves once per attempt. A second save raises unless you pass replace=True. Use it for a cumulative snapshot that grows while the block works: save the changed model again with replace=True. Only the thread that made the first save may replace it.
Has limits. The value may be at most 5 MiB, measured as the UTF-8 length of its compact JSON, integers beyond ±2^53 are refused, because the engine parses JSON with JavaScript numbers, and so are NaN and infinity, which JSON has no form for, and text that is not valid UTF-8, such as a lone surrogate, or that holds a NUL character, which the engine’s database cannot store. The check reads the value before any JSON-only serializer runs, so a @field_serializer(..., when_used="json") that maps NaN away does not get it through: store None (or a finite number) in the field instead.
Raises on a failed save (wrong type, too large, invalid) and leaves the earlier snapshot untouched. Like any exception that means side effects are assumed. Because the save comes before every write, catching the error and returning side_effects=False is honest.
Survives a failure. The worker sends the saved value with whatever result the EXECUTE job ends with, also after an exception in run().
Belongs to run(). Calling it inside rollback() raises.

Branches run in parallel save once

A block that works on several devices through run_parallel collects the per-device snapshots first, saves them once as one model, and only then starts writing. Branches that each save, or each replace, would race for the one snapshot of the attempt.

The engine checks the value against the registered schema and drops a snapshot that does not fit; the result itself is never lost. Rollback data is left out of the worker’s job log and of exception messages, but it is stored unencrypted by the engine: a snapshot of a device configuration is as sensitive as the configuration.

The block decides

forward.rollback_data is None when no snapshot was saved, when the engine rejected it (forward.rollback_data_rejected says why), when the outcome of the attempt is unknown (forward.outcome_known is False), or when the engine’s retention removed it after the run settled. What the rollback does then is up to the block: restore from params and context if it can, or refuse. It also decides what to do when it runs a second time, which forward.previous_rollbacks shows. A refusal is a failed FunctionBlockRollbackResult; the entry ends ROLLBACK_FAILED and an operator re-runs or resolves it.

When designing for rollback, consider:

Save before the first write, so the snapshot is the state the rollback must restore.
Idempotent rollbacks — a rollback may run again after an earlier one failed or was lost; check previous_rollbacks and the device state rather than assuming a clean start.
Partial rollback — if your block made three changes and only two succeeded, rollback should handle the partial state.

When to Split Function Blocks

A function block should encapsulate a single concern. Signs that a block is doing too much:

It opens connections to multiple devices sequentially.
It mixes read and write operations (e.g., fetching config then deploying a template).
It contains steps that could reasonably be retried independently.
Different parts of its logic have different purity or idempotency characteristics.

Splitting provides concrete benefits:

Benefit	Why it matters
Fine-grained retry	The workflow engine can retry a failed step without re-running successful ones
Accurate safety semantics	Each block can declare its own `is_pure` and `is_idempotent` flags
Workflow composition	Smaller blocks can be reused across multiple workflows
Clearer logging	Each block has its own execution log and result

Example: config deploy pipeline

Instead of one large ConfigDeploy block:

ConfigBackup (facts, is_pure=True) — read and store current config
ConfigValidate (check, is_pure=True) — validate the new template
ConfigPush (configure, is_idempotent=True) — deploy the template
ConfigVerify (check, is_pure=True) — verify post-deploy state

If step 3 fails, the engine can retry it without re-running steps 1 and 2.

Error Handling Strategies

Function blocks can signal failure in two ways:

1. Return `success=False`

The preferred approach. Return a FunctionBlockResult with success=False and a descriptive message:

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

The workflow engine receives the result and decides how to proceed (retry, skip, abort) based on the workflow configuration.

A failed job of a block registered is_pure or is_idempotent, or one whose result says side_effects=False, is retried by the engine, up to three times. When the failure would come out the same on every attempt (a verdict about the input or the network, not a fault), add retryable=False so the engine fails the step on its first attempt instead:

return FunctionBlockResult(
    message=f"Could not discover the named host(s): {missing}.",
    success=False,
    data=data,
    retryable=False,
)

The worker does this itself when a job’s parameters fail validation. An exception raised by the block stays retryable, and always means side effects are assumed. If the failure comes from a check that ran before anything was sent, also add side_effects=False (see Side effects of one attempt).

2. Raise an exception

If your code raises an unhandled exception, the execute_function_block() wrapper catches it and converts it to a failed result automatically:

# This exception becomes:
# FunctionBlockResult(message="Error executing function block: ...", success=False)
raise ConnectionError("SSH connection timed out")

The propagate_exceptions parameter on execute_function_block() controls this behavior:

Value	Behavior	Use case
`False` (default)	Exceptions are caught and converted to failed results	Production
`True`	Exceptions propagate with full stack traces	Testing and debugging

The testing framework sets propagate_exceptions=True by default so that pytest shows the actual error location rather than a generic failure message.

Prefer explicit failure over exceptions

Returning success=False with a clear message gives the workflow engine and operators more actionable information than a raw exception. Reserve exceptions for truly unexpected situations.

2026-02-26 17:18:082026-11-06 16:23:18