rs_workflows/utils/prefect.md

<< Back to index

Utilities for working with Prefect (run logger and variables).

get_logger(name=None)

Return the Prefect run logger when called from a Prefect flow or task run, otherwise a standard logger (e.g. when the workflow code is used outside of Prefect).

Parameters:

Name	Type	Description	Default
name	str | None	name of the standard logger. Default: module name of the caller.	None

Source code in docs/rs-client-libraries/rs_workflows/utils/prefect.py

def get_logger(name: str | None = None) -> logging.Logger | logging.LoggerAdapter:
    """
    Return the Prefect run logger when called from a Prefect flow or task run,
    otherwise a standard logger (e.g. when the workflow code is used outside of Prefect).

    Args:
        name: name of the standard logger. Default: module name of the caller.
    """
    try:
        return get_run_logger()
    except MissingContextError:
        if name is None:
            frame = inspect.currentframe()
            name = frame.f_back.f_globals.get("__name__", "rspy") if frame and frame.f_back else "rspy"
        return logging.getLogger(name)

update_prefect_variable(variable_name, updates) async

Merge updates into a Prefect variable and verify that they were persisted.

Source code in docs/rs-client-libraries/rs_workflows/utils/prefect.py

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_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