parent_path = Path(__file__).resolve().parent
sys.path.append(os.path.relpath(parent_path))

.
├── flow.py
├── functions
│   ├── function_mapping
│   │   ├── function_a.py
│   │   └── function_b.py
│   └── function_transform
│       ├── function_c.py
│       └── function_d.py
└── task
    ├── mapping
    │   └── task_a.py
    └── transform
        ├── task_b.py
        └── task_c.py

# in flow.py
from task.mapping.task_a import task_a

task

# in task_a.py
from functions.function_mapping.function_a import function_a

ModuleNotFoundError: No module named 'functions'

functions

start_date = Parameter("start_date", default=pendulum.now("Europe/Kiev").add(days=-1).to_date_string())
start_date = Parameter("start_date", default=(date.today() - timedelta(1)).strftime("%Y-%m-%d"))

2021-10-14

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