Here's how you can implement dynamic flow names based on function inputs in Prefect 3.x. I'll show you two approaches: 1. Using a callable for the flow run name:

python
from prefect import flow
from datetime import datetime

def generate_flow_name(parameters):
    return f"process_data_{parameters['data_type']}_{datetime.now().strftime('%Y%m%d')}"

@flow(flow_run_name=generate_flow_name)
def process_data(data_type: str, value: int):
    print(f"Processing {data_type} with value {value}")

# When called, this will create a flow run name like "process_data_sales_20231215"
process_data("sales", 42)

2. Using an f-string directly:

python
from prefect import flow

@flow(flow_run_name="{parameters[data_type]}_processing_{date:%Y%m%d}")
def process_data(data_type: str, value: int):
    print(f"Processing {data_type} with value {value}")

# When called, this will create a flow run name like "sales_processing_20231215"
process_data("sales", 42)

A few important notes: 1. The

flow_run_name

parameter can access: - `parameters`: a dictionary of the flow's input parameters - `date`: the current datetime object - `tags`: any tags applied to the flow run - `flow_name`: the original name of the flow 2. If you need more complex logic, using a callable (first example) gives you more flexibility to format the name exactly as needed. 3. The flow name itself (as defined by the function name) remains static - what we're actually modifying is the flow run name, which is what you'll see in the Prefect UI and logs. This feature is particularly useful when you need to: - Track different runs based on input parameters - Create more meaningful names for monitoring and debugging - Generate unique identifiers for each flow run - Organize flow runs based on their inputs or purpose Remember that while the flow run name can be dynamic, it should still result in a valid string that can be used as an identifier in your system.

Actually, you don't need to set both - let me explain why and show you the best practices: 1. Flow Name: - This is automatically derived from your function name by default - Can be overridden using the

name

parameter in the decorator if needed - Should be relatively static as it identifies the flow definition itself 2. Flow Run Name: - This is what you want to make dynamic - Represents specific executions of your flow - Can include runtime information like parameters, timestamps, etc. Here's the recommended approach:

python
from prefect import flow
from datetime import datetime

# Option 1: Just use flow_run_name (recommended for most cases)
@flow(flow_run_name="{parameters[data_type]}_{date:%Y%m%d}")
def process_data(data_type: str, value: int):
    return f"Processing {data_type} with value {value}"

# Option 2: If you really need both custom flow name and dynamic run names
@flow(
    name="data_processor",  # Static flow name
    flow_run_name="{parameters[data_type]}_{date:%Y%m%d}"  # Dynamic run name
)
def process_data(data_type: str, value: int):
    return f"Processing {data_type} with value {value}"

# When you run either version:
result = process_data("sales", 42)
# Flow name will be "process_data" (Option 1) or "data_processor" (Option 2)
# Flow run name will be something like "sales_20231215"

You can also update the flow run name dynamically during execution:

python
from prefect import flow
from prefect.context import get_run_context

@flow
def process_data(data_type: str, value: int):
    # Get the current context and update the flow run name
    context = get_run_context()
    context.flow_run.name = f"{data_type}_processing_{datetime.now().strftime('%Y%m%d')}"
    return f"Processing {data_type} with value {value}"

Key points to remember: 1. The flow name should be stable and used to identify your flow definition 2. The flow run name can (and often should) be dynamic to help identify specific executions 3. In most cases, you only need to set the

flow_run_name

parameter 4. Only set a custom

name

if you have a specific reason to override the default function name The distinction is important because: - Flow names are used for flow registration and retrieval - Flow run names are used for monitoring and tracking specific executions - You'll see both in the UI, but they serve different purposes

Yes! When using a callable for

flow_run_name

, the function receives a dictionary with several useful context variables. Here's how you can use them:

python
from prefect import flow
from datetime import datetime

def generate_flow_run_name(context: dict):
    """
    Available context keys:
    - parameters: Dictionary of flow parameters
    - flow_name: Original name of the flow
    - date: Current datetime object
    - tags: List of tags applied to the flow run
    """
    params = context["parameters"]
    flow_name = context["flow_name"]
    current_date = context["date"]
    
    return f"{flow_name}_{params['data_type']}_{current_date:%Y%m%d}"

@flow(flow_run_name=generate_flow_run_name)
def process_data(data_type: str, value: int, custom_prefix: str = None):
    return f"Processing {data_type} with value {value}"

# Usage
process_data("sales", 42)  # Creates run name like: "process_data_sales_20231215"

You can get even more creative with the context:

python
from prefect import flow
from datetime import datetime

def generate_flow_run_name(context: dict):
    params = context["parameters"]
    
    # Add custom prefix if provided
    prefix = params.get("custom_prefix", "default")
    
    # Create a formatted timestamp
    timestamp = context["date"].strftime("%Y%m%d_%H%M")
    
    # Include the data type and value in the name
    return f"{prefix}_{params['data_type']}_{params['value']}_{timestamp}"

@flow(flow_run_name=generate_flow_run_name)
def process_data(data_type: str, value: int, custom_prefix: str = None):
    return f"Processing {data_type} with value {value}"

# Different ways to use it
process_data("sales", 42, custom_prefix="PROD")  # Creates: "PROD_sales_42_20231215_1430"
process_data("inventory", 100)  # Creates: "default_inventory_100_20231215_1430"

You can also combine this with runtime updates:

python
from prefect import flow
from prefect.context import get_run_context

def generate_initial_name(context: dict):
    return f"INITIAL_{context['parameters']['data_type']}"

@flow(flow_run_name=generate_initial_name)
def process_data(data_type: str, value: int):
    # Get the current context
    context = get_run_context()
    
    # Update the name during execution if needed
    if value > 100:
        context.flow_run.name = f"HIGH_VALUE_{data_type}_{value}"
    else:
        context.flow_run.name = f"NORMAL_{data_type}_{value}"
        
    return f"Processing {data_type} with value {value}"

Key points to remember: 1. The generator function receives a context dictionary with useful information: - `parameters`: All flow input parameters - `flow_name`: The original flow name - `date`: Current datetime object - `tags`: Any tags applied to the flow run 2. You can access any flow parameters through

context["parameters"]

3. The function must return a string that will be used as the flow run name 4. You can combine this with runtime updates using

get_run_context()

if you need to change the name during flow execution This approach gives you a lot of flexibility in naming your flow runs based on both input parameters and runtime conditions.