create_flow_run

import platform
import prefect
from prefect import Flow, Parameter, task


CHILD_FLOW_NAME = "child_flow_example"


@task(log_stdout=True)
def hello_world_child(x: str):
    print(f"Hello {x} from {CHILD_FLOW_NAME}!")
    print(
        f"Running this task with Prefect: {prefect.__version__} and Python {platform.python_version()}"
    )


with Flow(CHILD_FLOW_NAME) as flow:
    user_input = Parameter("user_input", default="Marvin")
    hw = hello_world_child(user_input)

import os
import subprocess
from prefect import Flow, task, Parameter
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run

PARENT_FLOW_NAME = "parent_flow_example"
CHILD_FLOW_NAME = "child_flow_example"
PREFECT_PROJECT_NAME = "community"


@task(log_stdout=True)
def hello_world_parent():
    print(f"Hello from {PARENT_FLOW_NAME}!")
    return PARENT_FLOW_NAME


with Flow(PARENT_FLOW_NAME) as parent_flow:
    normal_non_subflow_task = hello_world_parent()
    param = Parameter("user_name", default="world")
    first_child_flow_run_id = create_flow_run(
        flow_name=CHILD_FLOW_NAME,
        project_name=PREFECT_PROJECT_NAME,
        parameters=dict(user_input=param),
        task_args=dict(name="First subflow"),
        upstream_tasks=[normal_non_subflow_task],
    )
    first_child_flowrunview = wait_for_flow_run(
        first_child_flow_run_id,
        raise_final_state=True,
        stream_logs=True,
        task_args=dict(name="Wait for the first subflow"),
    )

if __name__ == "__main__":
    dir_path = os.path.dirname(os.path.realpath(__file__))
    subprocess.run(
        f"prefect register --project {PREFECT_PROJECT_NAME} -p {dir_path}", shell=True,
    )
    subprocess.run(
        f"prefect run --name {PARENT_FLOW_NAME} --project {PREFECT_PROJECT_NAME}",
        shell=True,
    )
    subprocess.run(
        f"prefect agent local start", shell=True,
    )

create_flow_run

create_flow_run

.map

create_flow_run

.map

start_date

end_date

create_flow_run

idempotency_key

raise_final_state

raise_final_state=True

with Flow("Mongo->BigQuery") as flow:
    min_time = DateTimeParameter(
        "min_time",
        required=False,
        default=(datetime.utcnow() - timedelta(hours=1)).replace(
            minute=0, second=0, microsecond=0
        ),
    )
    max_time = DateTimeParameter(
        "max_time",
        required=False,
        default=(datetime.utcnow() - timedelta(hours=1)).replace(
            minute=59, second=59, microsecond=999999
        ),
    )
    mongo_server_tag = Parameter("mongo_server_tag", required=True)
    mongo_url = Parameter("mongo_url", required=True)
    mongo_database = Parameter("mongo_database", required=True)
    mongo_database_subtask = Parameter("mongo_database_subtask", required=True)

    # Create a mapping of tasks for all collections in main Mongo database
    params = MONGO_COLLECTIONS["main"]
    MongoCollectionToBigQuery(
        task_run_name="{mongo_collection}",
        bq_project_id=prefect.context.get("BQ_PROJECT_ID"),
        bq_dataset_id=prefect.context.get("BQ_DATASET_ID"),
        bq_location=prefect.context.get("BQ_LOCATION"),
    ).map(
        mongo_collection=[i["mongo_collection"] for i in params],
        mongo_projection=[i.get("mongo_projection", None) for i in params],
        mongo_created_key=[i.get("mongo_created_key", None) for i in params],
        mongo_updated_key=[i.get("mongo_updated_key", None) for i in params],
        bq_collection_name=[i.get("bq_collection_name") for i in params],
        bq_stream_chunk_size=[i.get("bq_stream_chunk_size", 3000) for i in params],
        min_time=unmapped(min_time),
        max_time=unmapped(max_time),
        mongo_server_tag=unmapped(mongo_server_tag),
        mongo_url=unmapped(mongo_url),
        mongo_database=unmapped(mongo_database),
        task_args=dict(
            max_retries=3,
            retry_delay=timedelta(minutes=1)
        )
    )

    # Create a mapping of tasks for all collections in SubTask Mongo database
    params = MONGO_COLLECTIONS["subtask"]
    MongoCollectionToBigQuery(
        task_run_name="{mongo_collection}",
        bq_project_id=prefect.context.get("BQ_PROJECT_ID"),
        bq_dataset_id=prefect.context.get("BQ_DATASET_ID"),
        bq_location=prefect.context.get("BQ_LOCATION"),
    ).map(
        mongo_collection=[i["mongo_collection"] for i in params],
        mongo_projection=[i.get("mongo_projection", None) for i in params],
        mongo_created_key=[i.get("mongo_created_key", "created") for i in params],
        mongo_updated_key=[i.get("mongo_updated_key", "updated") for i in params],
        bq_collection_name=[i.get("bq_collection_name") for i in params],
        bq_stream_chunk_size=[i.get("bq_stream_chunk_size", 3000) for i in params],
        min_time=unmapped(min_time),
        max_time=unmapped(max_time),
        mongo_server_tag=unmapped(mongo_server_tag),
        mongo_url=unmapped(mongo_url),
        mongo_database=unmapped(mongo_database_subtask),
        task_args=dict(
            max_retries=3,
            retry_delay=timedelta(minutes=1)
        )
    )

flow.executor = LocalDaskExecutor()