async def update_prefect_variable(variable_name: str, updates: dict[str, Any]) -> dict[str, Any]:
"""Merge updates into a Prefect variable and verify that they were persisted."""
logger = get_run_logger()
logger.info("Reading current Prefect variable %s", variable_name)
raw_value = await cast(Awaitable[Any], Variable.get(variable_name, default={}))
logger.info(
"Read Prefect variable %s: type=%s, keys=%s",
variable_name,
type(raw_value).__name__,
sorted(raw_value) if isinstance(raw_value, dict) else [],
)
value = _deep_merge(raw_value if isinstance(raw_value, dict) else {}, updates)
logger.info("Updating Prefect variable %s with keys=%s", variable_name, sorted(updates))
await cast(Awaitable[Any], Variable.set(variable_name, value, overwrite=True))
saved_value = await cast(Awaitable[Any], Variable.get(variable_name, default={}))
if not isinstance(saved_value, dict):
raise RuntimeError(
f"Prefect variable {variable_name!r} was not updated: expected a dictionary, "
f"got {type(saved_value).__name__}",
)
if not _contains_updates(saved_value, updates):
raise RuntimeError(f"Prefect variable {variable_name!r} was not updated: expected nested updates {updates!r}")
logger.info("Verified Prefect variable %s update for keys=%s", variable_name, sorted(updates))
return saved_value