A warp 9 Prefect Orion flow with tasks

warp_flow_with_tasks.py

from prefect import flow, task, get_run_logger


@task(name="Evasive maneuvers")
def evasive_maneuvers():
    logger = get_run_logger()
    logger.warning("Enemy vessel detected")
    logger.warning("Incoming photon torpedo")
    logger.error("Evasive maneuvers failed!")
    logger.debug("Warp drive temperature: 535°C")
    logger.critical("Fire detected in engine room!")


@task(name="Warm up warp drive")
def warm_up_warp_drive():
    logger = get_run_logger()
    logger.info("Preparing for warp 9")
    logger.debug("Warp drive temperature: 200°C")


@task(name="Engage warp 9")
def engage_warp_9():
    logger = get_run_logger()
    logger.info("Warp drive engaged")
    logger.info("Traveled 1,442,442 km at warp 9")
    logger.info("Warp drive disengaged")


@flow(name="Engage!")
def warp_flow():
    warm_up_warp_drive()
    evasive_maneuvers()
    engage_warp_9()


if __name__ == "__main__":
    warp_flow()