import logging
import pendulum
import prefect
from prefect import Flow, Parameter, task
from prefect.storage import Docker
from prefect.schedules import CronSchedule
from prefect.tasks.prefect import StartFlowRun


logger = logging.getLogger(__name__)

weekday_schedule = CronSchedule(
    '41 10 * * 1-5',
    start_date=pendulum.now(tz='Europe/London')
)


@task
def calculate_flow_end_date(end_date: str):
    if end_date is not None:
        return end_date
    return prefect.context.get('scheduled_start_time').to_date_string()


common_flow = StartFlowRun(
    flow_name='already_existing_flow1',
    project_name='my_project_name',
    wait=True,
)
baseline_flow = StartFlowRun(
    flow_name='already_existing_flow2',
    project_name='my_project_name',
    wait=True,
)

with Flow("my_scheduled_flow",
          schedule=weekday_schedule,
          storage=Docker(
              base_image='my-docker-image:latest',
              local_image=True,
          )) as flow:

    num_days_parameter = Parameter('num_days', default=1)
    num_back_fill_days_parameter = Parameter('num_back_fill_days', default=1)
    end_date_parameter = Parameter('end_date', default=None)

    task_end_date = calculate_flow_end_date(end_date_parameter)
    <http://logger.info|logger.info>('Task End Date: %s', task_end_date)
    <http://logger.info|logger.info>('Num Days: %s', num_days_parameter)
    <http://logger.info|logger.info>('Num Days Backfill: %s', num_back_fill_days_parameter)

    common_flow_result = common_flow(parameters={
        'num_days': num_days_parameter,
        'num_back_fill_days': num_back_fill_days_parameter,
        'end_date': task_end_date,
    })
    baseline_flow_result = baseline_flow(
        upstream_tasks=[common_flow_result],
        parameters={
            'num_days': num_days_parameter,
            'num_back_fill_days': num_back_fill_days_parameter,
            'end_date': task_end_date,
        }
    )

with Flow(
    'common_flow',
    storage=Docker(
        base_image='my-docker-image:latest',
        local_image=True,
    )) as flow:

    num_days_parameter = Parameter('num_days', required=True)
    num_back_fill_days_parameter = Parameter('num_back_fill_days', required=True)
    end_date_parameter = Parameter('end_date', required=True)

    served_ads_command = ShellTask(
        name='my_first_task',
        command=construct_etl_command(
            app_mode=settings.app_mode,
            table_name='task1',
            num_days_cli_arg='num_back_fill_days',
            num_days=num_back_fill_days_parameter,
            end_date=end_date_parameter,
            database=settings.database,
        ),
        stream_output=True,
    )

[2021-10-18 09:30:15+0000] INFO - prefect.served_advert_task | /tmp/prefect-ufbr5w5b: line 1: Parameter:: No such file or directory [2021-10-18 09:30:15+0000] ERROR - prefect.served_advert_task | Command failed with exit code 1

wait_for_flow_run

wait_for_flow_run

FlowRunView

is_failed()

prefect.context.get("scheduled_start_time")

CronSchedule(cron, start_date=pendulum.datetime(2021, 1, 1, tz=tz))

with Flow(
    name="horse_racing_data",
) as flow:
    dates = get_dates_task(days_back=days_back, days_ahead=days_ahead, dt_format="%d-%b-%Y")

    raw_sectional_data = apply_map(get_puntingform_sectional_dump_task, date=dates)


    spell_stats_data = apply_map(
        query_db_for_df_task, path_to_sql=unmapped("sql/select_spell_count.sql")
    )
    
    enriched_pf_data = apply_map(
        calculate_runners_spell_stats_task, pf_sectional_df=raw_sectional_data, spell_data=spell_stats_data
    )

apply_map

calculate_runners_spell_stats_task

raw_sectional_data

spell_stats_data

prefect.core.flow.Flow.serialized_hash

register

StartFlowRun.run()

Error during execution of task: ClientError([{'message': '[{\'extensions\': {\'path\': \'$.selectionSet.insert_task_run_artifact.args.objects\', \'code\': \'constraint-violation\'}, \'message\': \'Not-NULL violation. null value in column "tenant_id" violates not-null constraint\'}]', 'locations': [{'line': 2, 'column': 5}], 'path': ['create_task_run_artifact'], 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': '[{\'extensions\': {\'path\': \'$.selectionSet.insert_task_run_artifact.args.objects\', \'code\': \'constraint-violation\'}, \'message\': \'Not-NULL violation. null value in column "tenant_id" violates not-null constraint\'}]'}}}])

0.14.22

server

AZURE_STORAGE_CONNECTION_STRING

run_config

with Flow('test_flow',executor=LocalExecutor()) as flow:

df = sometask_that_returns_a_df()

count_of_records_in_df = len(df.index)

execute_task_2(count_of_records_in_df)

KubernetesRun

run_config=KubernetesRun(
                env={
                    "AWS_ACCESS_KEY_ID": AWS_ACCESS_KEY_ID,
                    "AWS_SECRET_ACCESS_KEY": AWS_SECRET_ACCESS_KEY
                })

Error downloading Flow from S3: An error occurred (InvalidAccessKeyId) when calling the GetObject operation: The AWS Access Key Id you provided does not exist in our records.