def METHOD_NAME(
    airflow_client: AirflowClient,
    new_snapshots: t.List[Snapshot],
    environment: Environment,
    is_dev: t.Optional[bool] = None,
) -> None:
    if is_dev is None:
        is_dev = environment.name != c.PROD
    plan_request_id = random_id()
    airflow_client.apply_plan(new_snapshots, environment, plan_request_id, is_dev=is_dev)
    plan_application_dag_id = common.plan_application_dag_id(environment.name, plan_request_id)
    plan_application_dag_run_id = airflow_client.wait_for_first_dag_run(
        plan_application_dag_id, DAG_CREATION_WAIT_INTERVAL, DAG_CREATION_RETRY_ATTEMPTS
    )
    assert airflow_client.wait_for_dag_run_completion(
        plan_application_dag_id, plan_application_dag_run_id, DAG_RUN_POLL_INTERVAL
    )