Great question. In Prefect 3.x you can read the current deployment’s name at runtime and branch your logic. Two good options: 1) Use the deployment name inside the flow (quick and minimal change)

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.runtime import deployment

@flow(
    name="EBAY",
    flow_run_name="{seller_id}-{data_cycle}",
    log_prints=True,
    task_runner=ThreadPoolTaskRunner(max_workers=8),
    on_failure=[pipeline_alert], on_crashed=[pipeline_alert],
    on_running=[pipeline_alert], on_completion=[pipeline_alert], on_cancellation=[pipeline_alert],
)
def run(seller_id: str, selling_partner_id: str, data_cycle: str):
    dname = (deployment.name or "").lower()  # '' if not running from a deployment
    # Simple mapping based on substrings in the deployment name
    if "prod" in dname:
        period = 7
    elif "stage" in dname or "staging" in dname:
        period = 3
    else:
        period = 1  # default for dev / local

    ebay_extraction = EbayExtraction(
        seller_id, selling_partner_id, data_cycle, extraction_period=period
    )
    # ...

2) Recommended: make

extraction_period

a flow parameter and set it per-deployment This is more explicit and easier to manage at scale.

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(
    name="EBAY",
    flow_run_name="{seller_id}-{data_cycle}",
    log_prints=True,
    task_runner=ThreadPoolTaskRunner(max_workers=8),
    on_failure=[pipeline_alert], on_crashed=[pipeline_alert],
    on_running=[pipeline_alert], on_completion=[pipeline_alert], on_cancellation=[pipeline_alert],
)
def run(
    seller_id: str,
    selling_partner_id: str,
    data_cycle: str,
    extraction_period: int = 7,  # default
):
    ebay_extraction = EbayExtraction(
        seller_id, selling_partner_id, data_cycle, extraction_period=extraction_period
    )
    # ...

if __name__ == "__main__":
    # Create deployments with different defaults
    run.deploy(
        name="ebay-dev",
        work_pool_name="your-ecs-ec2-pool",
        parameters={"extraction_period": 1},
        tags=["dev"],
    )
    run.deploy(
        name="ebay-stage",
        work_pool_name="your-ecs-ec2-pool",
        parameters={"extraction_period": 3},
        tags=["stage"],
    )
    run.deploy(
        name="ebay-prod",
        work_pool_name="your-ecs-ec2-pool",
        parameters={"extraction_period": 7},
        tags=["prod"],
    )

Notes: -

prefect.runtime.deployment.name

returns a string of the current deployment’s name, or

''

when not running from a deployment. Docs: Runtime context. - Parameters set via

deploy(..., parameters={...})

become the defaults for runs from that deployment. Docs: Deployments. If you share how you’re naming your deployments (e.g., EBAY-dev / EBAY-stage / EBAY-prod), I can tailor the mapping logic exactly to your scheme.