...

dev_stage = os.environ["DEV_STAGE"]

if dev_stage.lower() == "prd":
    handler = slack_notifier(only_states=[Failed])  # we can call it early
else:
    handler = slack_notifier(ignore_states=[Pending, Running, Finished])


sync_airbyte_connection = AirbyteConnectionTask(
    max_retries=3, retry_delay=timedelta(seconds=10)
)

with Flow(
    f"{dev_stage} - Airbyte - Flow",
    run_config=UniversalRun(
        labels=["airbyte", dev_stage], env={"DEV_STAGE": dev_stage}
    ),
    executor=LocalDaskExecutor(),
    state_handlers=[handler],
) as flow:

    # Airbyte connection strings
    airbyte_connections = [
        "xxx",
       ...
    ]

    airbyte_sync = sync_airbyte_connection.map(
        airbyte_server_host=unmapped(PrefectSecret("AIRBYTE_PUBLIC_IP")),
        airbyte_server_port=unmapped(8000),
        airbyte_api_version=unmapped("v1"),
        connection_id=airbyte_connections,
    )

if dev_stage == 'prd':
    flow.storage = GitHub(
        repo="OrthoFi/data_warehouse_pipeline",  # name of repo
        path="containers/prefect/src/flows/sub_flows/flow_airbyte.py",  # location of flow file in repo
        access_token_secret="GITHUB_ACCESS_TOKEN",  # name of personal access token secret
    )
else:
    flow.storage = GitHub(
        repo="OrthoFi/data_warehouse_pipeline",  # name of repo
        path="containers/prefect/src/flows/sub_flows/flow_airbyte.py",  # location of flow file in repo
        access_token_secret="GITHUB_ACCESS_TOKEN",  # name of personal access token secret
        ref='development'
    )