See my blog post HomeAssistant: Gracefully detecting and retrying on errors in YAML scripts for background and pure-YAML approximations.
In particular the python_script are the following:
Python script files
python_scripts/services.yaml
actions_with_error_state:
name: Call Actions returning Error State
icon: mdi:alert-circle-check-outline
description: >-
Invoke the given action list, *returning* whether the given action list
succeeded or information about the failing error.
Return structure:
* `ok` (boolean): `true` if all actions succeeded, `false` otherwise
* `error` (`None` or object): If an action failed, information about the error
* `action` (string): Name of the failing action
* `idx` (integer): 0-based index of the failing action in the list of actions
* `exc` (exception): Caught service exception, can be stringified to get error message
fields:
actions:
name: Actions to execute
selector:
action: {}
required: true
actions_with_retries:
name: Call Actions and Restart on Error
icon: mdi:repeat-variant
description: >-
Repeats the given action steps until they either succeed or the number of
retries has been exhausted.
Return structure:
* `ok` (boolean): `true` if all actions succeeded at some point, `false` otherwise
* `retries` (integer): How often the actions were retried before succeeding (in range 0 to `retries`)
* `error` (`None` or object): If an action failed, information about the error
* `action` (string): Name of the failing action
* `idx` (integer): 0-based index of the failing action in the list of actions
* `exc` (exception): Caught service exception, can be stringified to get error message
If `fail_on_error` is true, this action will fail and no value is returned.
fields:
actions:
name: Actions to execute
selector:
action: {}
required: true
retries:
name: Number of Retries
description: >-
How often to retry *all* given actions before returning the first
encountered error.
selector:
number:
min: 1
default: 3
delay:
name: Retry Delay
description: >-
Delay between retries of the action list.
selector:
duration:
enable_millisecond: true
default:
hours: 0
minutes: 0
seconds: 1
milliseconds: 0
fail_on_error:
name: Fail on Too Many Retries
description: >-
Note that you must still set a “Response Variable” when using this
option or reported errors will not cause the calling script to exit.
selector:
boolean: {}
default: false
python_scripts/actions_with_error_state.py
actions = data["actions"] # Required action list
for idx, action in enumerate(actions):
domain, name = action["action"].split(".", 1) # "light.turn_on"
payload = action.get("data", {})
payload.update(action.get("target", {}))
try:
logger.info(f"Running action “{domain}.{name}”")
hass.services.call(domain, name, payload, blocking=True)
except Exception as exc:
logger.warning(f"Action “{domain}.{name}” failed: {exc}")
output["ok"] = False
output["error"] = {"action": action["action"], "idx": idx, "exc": exc}
break
else:
output["ok"] = True
output["error"] = None
python_scripts/actions_with_retries.py
actions = data["actions"] # Required action list
retries = data.get("retries", 3)
delay = data.get("delay", 1)
fail_on_error = data.get("fail_on_error", False)
# Convert delay to fraction of second without violating sandbox
try:
delay = float(delay)
except TypeError:
delay = datetime.timedelta(
hours = delay.get("hours", 0),
minutes = delay.get("minutes", 0),
seconds = delay.get("seconds", 0),
milliseconds = delay.get("milliseconds", 0),
).total_seconds()
first_error = None
for retry in range(retries):
# Call `actions_with_error_state` to perform failable action run
result = hass.services.call(
"python_script", "actions_with_error_state",
{ "actions": actions },
blocking=True,
return_response=True,
)
if result["ok"]:
# Forward success with added retry count
output.update(result)
output["retries"] = retry
break
# Save first error to forward it in case of ultimate failure
if first_error is None:
first_error = result
# Apply delay only if not the last iteration
if retry < retries - 1:
time.sleep(delay)
# Handle if `break` was not reached / we are out of retries
else:
# Forward first encountered error either as exception (“action failure”)
# or structured valued interpretable by caller
if fail_on_error:
raise first_error["error"]["exc"]
else:
output.update(first_error)
output["retries"] = retries
Usage
Example usage would be:
alias: "Door Bell: Send and Dismiss Notification (Background Task)"
icon: mdi:alarm-bell
sequence:
- alias: "Ntfy: Publish Notification (with Retries)"
action: python_script.actions_with_retries
data:
actions:
- action: ntfy.publish
data:
priority: "5"
sequence_id: doorbell
tags:
- bell
title: Doorbell
message: Doorbell button has been pressed
target:
device_id: be3fc64c497985de2d8705e856a9d3fb
fail_on_error: true
response_variable: result # Unused variable, but property is required for failure to be reported
- delay:
hours: 0
minutes: 1
seconds: 0
milliseconds: 0
- alias: "Ntfy: Dismiss Notification (with Retries)"
action: python_script.actions_with_retries
data:
actions:
- action: ntfy.clear
target:
device_id: be3fc64c497985de2d8705e856a9d3fb
data:
sequence_id: doorbell
enabled: true
delay:
hours: 0
minutes: 0
seconds: 1
milliseconds: 0
retries: 3
fail_on_error: true
response_variable: result # Unused variable, but property is required for failure to be reported
mode: restart
See the blog post for more details!