PREFECT_RUNNER_PROCESS_LIMIT

serv

PREFECT_LOGGING_LEVEL=DEBUG

serv

asyncio.wait

timeout

asyncio.wait

python
   import asyncio

   async def serve_wrapper(task):
       # Your task logic here
       pass

   async def main(tasks):
       # Add a timeout value in seconds
       timeout_value = 10
       # Wait for tasks with timeout
       done, pending = await asyncio.wait(
           [asyncio.create_task(serve_wrapper(task)) for task in tasks],
           timeout=timeout_value
       )

       # Handle completed tasks
       for task in done:
           try:
               result = await task
               print(f"Task completed with result: {result}")
           except Exception as e:
               print(f"Task failed with exception: {e}")

       # Handle tasks that did not complete within the timeout
       for task in pending:
           print(f"Task did not complete within the timeout: {task}")

   # Example usage
   asyncio.run(main(your_task_list))

done

pending

timeout_seconds

time.sleep(30)

asyncio.sleep

time.sleep

await asyncio.sleep

PREFECT_ENHANCED_CANCELLATION=True

python
from prefect import flow, get_run_logger
import asyncio

@flow(timeout_seconds=5)
async def testing_flow():
    # Assuming common.load_environment() is a synchronous call
    common.load_environment()
    logger = get_run_logger()
    <http://logger.info|logger.info>('Starting second spectrum games fetch....')
    
    await asyncio.sleep(30)  # Use asyncio.sleep instead of time.sleep

    <http://logger.info|logger.info>('Finished second spectrum games fetch....')

if __name__ == "__main__":
    asyncio.run(testing_flow())

PREFECT_ENHANCED_CANCELLATION=True

PREFECT_ENHANCED_CANCELLATION

PREFECT_EXPERIMENTAL_ENABLE_ENHANCED_CANCELLATION

true

prefect flow-run ls

serv

prefect deployment inspect

prefect flow-run ls

PREFECT_LOGGING_INTERNAL_LEVEL

ERROR

ERROR

WARNING

INFO

DEBUG

timeout_seconds

python
from prefect import flow, get_run_logger
import asyncio

@flow(timeout_seconds=5)
async def long_running_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Starting long running task...")
    
    await asyncio.sleep(30)  # Simulate a long-running operation

    <http://logger.info|logger.info>("Finished long running task...")

if __name__ == "__main__":
    asyncio.run(long_running_flow())

long_running_flow

timeout_seconds=5

asyncio.sleep(30)

prefect flow-run cancel

bash
   prefect flow-run ls

bash
   prefect flow-run cancel <FLOW_RUN_ID>

<FLOW_RUN_ID>

PREFECT_ENHANCED_CANCELLATION

get_client

python
import asyncio
from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState

async def cancel_flow_run(flow_run_id):
    async with get_client() as client:
        # Transition the state of the flow run to 'Cancelled'
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state="Cancelled"
        )
        print(f"Flow run {flow_run_id} has been cancelled.")

# Replace <FLOW_RUN_ID> with the actual flow run ID you want to cancel
flow_run_id = "<FLOW_RUN_ID>"

if __name__ == "__main__":
    asyncio.run(cancel_flow_run(flow_run_id))

get_client

set_flow_run_state

Cancelled

AttributeError

python
import asyncio
from prefect import get_client
from prefect.orion.schemas.states import Cancelled

async def cancel_flow_run(flow_run_id):
    async with get_client() as client:
        # Create a Cancelled state instance instead of using a string
        state = Cancelled()
        # Transition the state of the flow run to 'Cancelled'
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state=state
        )
        print(f"Flow run {flow_run_id} has been cancelled.")

# Replace <FLOW_RUN_ID> with the actual flow run ID you want to cancel
flow_run_id = "<FLOW_RUN_ID>"

if __name__ == "__main__":
    asyncio.run(cancel_flow_run(flow_run_id))

Cancelled

prefect.orion.schemas.states

get_client()

AttributeError

python
import asyncio
from prefect import get_client
from prefect.orion.schemas.states import StateType, State

async def cancel_flow_run(flow_run_id):
    async with get_client() as client:
        # Create a state object representing the Cancelled state
        cancelled_state = State(type=StateType.CANCELLED, name="Cancelled")

        # Transition the state of the flow run to 'Cancelled'
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state=cancelled_state
        )
        print(f"Flow run {flow_run_id} has been cancelled.")

# Replace <FLOW_RUN_ID> with the actual flow run ID you want to cancel
flow_run_id = "<FLOW_RUN_ID>"

if __name__ == "__main__":
    asyncio.run(cancel_flow_run(flow_run_id))

State

StateType.CANCELLED

flow_run_id

python
import asyncio
from prefect.client import get_client
from prefect import flow
from prefect.orion.schemas.states import State, StateType

async def cancel_flow_run(flow_run_id):
    async with get_client() as client:
        # Create a State instance to represent the cancellation
        cancelled_state = State(type=StateType.CANCELLED)

        # Set the flow run's state to "Cancelled"
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state=cancelled_state
        )
        print(f"Flow run {flow_run_id} has been cancelled.")

# Replace <FLOW_RUN_ID> with the actual flow run ID to cancel
flow_run_id = "<FLOW_RUN_ID>"

if __name__ == "__main__":
    asyncio.run(cancel_flow_run(flow_run_id))

State

StateType.CANCELLED

serv

python
from prefect import flow, task
from prefect.deployments import Deployment

@task
def my_task():
    print("Running my task...")

@flow
def my_flow():
    my_task()

if __name__ == "__main__":
    # Create a deployment for my_flow that runs every 15 seconds
    deployment = Deployment.build_from_flow(
        flow=my_flow,
        name="15-second-interval",
        # Set an interval of 15 seconds
        schedule=IntervalSchedule(interval=timedelta(seconds=15))
    )
    
    # Serve the deployment
    deployment.serve()

@flow

@task

Deployment.build_from_flow

IntervalSchedule

deployment.serve()

rrule

rrule

python
from prefect import flow, task
from prefect.deployments import Deployment
from prefect.schedules import RRuleSchedule
from dateutil.rrule import rrulestr

@task
def my_task():
    print("Executing task...")

@flow
def my_flow():
    my_task()

if __name__ == "__main__":
    # Define an RRule for running every 15 seconds
    rrule = rrulestr("FREQ=SECONDLY;INTERVAL=15")

    # Create a deployment with an RRule schedule
    deployment = Deployment.build_from_flow(
        flow=my_flow,
        name="rrule-interval-15-seconds",
        schedule=RRuleSchedule(rrule=rrule)
    )
    
    # Serve the deployment
    deployment.serve()

RRuleSchedule

rrule

dateutil.rrule

rrule

FREQ=SECONDLY;INTERVAL=15

RRuleSchedule

timezone

rrule

PREFECT_LOGGING_MARKUP: true

PREFECT_LOGGING_MARKUP=True

python
import os
from prefect import flow, get_run_logger

# Ensure this environment variable is set for markup usage
os.environ["PREFECT_LOGGING_MARKUP"] = "true"

@flow
def my_flow():
    logger = get_run_logger()
    # Using Rich markup to style log message
    <http://logger.info|logger.info>("This is [bold red]fancy[/]")

if __name__ == "__main__":
    my_flow()

[bold red]fancy[/]

PREFECT_LOGGING_MARKUP

true

[bold bright_magenta]

[/]

pipeline

process_id

message_context

scheduled_date

new_flow_run.name

common.get_flow_run_url(new_flow_run)

python
import os
from prefect import flow, get_run_logger

# Ensure this environment variable is set for markup usage
os.environ["PREFECT_LOGGING_MARKUP"] = "true"

@flow
def example_flow():
    logger = get_run_logger()
    pipeline = "MyPipeline"
    process_id = 123
    message_context = "all systems go"
    scheduled_date = "2023-10-15"
    new_flow_run = type('FlowRun', (object,), {"name": "sample-flow"})()  # Mocking a new flow run object
    # Simulating a get_flow_run_url function
    common = type('Common', (object,), {"get_flow_run_url": lambda x: "<http://example.com/flow/>" + x.name})()
    
    # Logging with Rich markup style
    <http://logger.info|logger.info>(f'[bold bright_magenta]{pipeline} Pipeline run for process_id {process_id}, '
                f'message {message_context} scheduled for {scheduled_date} - '
                f'({new_flow_run.name} - {common.get_flow_run_url(new_flow_run)} )[/]')

if __name__ == "__main__":
    example_flow()

PREFECT_LOGGING_MARKUP

true

"Pipeline run for process_id: {process_id}, message: {message_context}, scheduled for: {scheduled_date}"

***

>>>

!!!

"** Important: Pipeline {pipeline} run scheduled **"

"Detailed logs available at: /path/to/detailed-log.txt"