Handle scheduled start time in local runs (plus tests) #1418

jlowin · 2019-08-28T14:59:23Z

Thanks for contributing to Prefect!

Please describe your work and make sure your PR:

adds new tests (if appropriate)
updates CHANGELOG.md (if appropriate)
updates docstrings for any new functions or function arguments, including docs/outline.toml for API reference docs (if appropriate)

Note that your PR will not be reviewed unless all three boxes are checked.

What does this PR change?

Properly handles scheduled_start_time for local flow.run() execution, including providing it for any retries.

docs/guide/core_concepts/flows.md

src/prefect/core/flow.py

codecov · 2019-08-29T00:57:01Z

Codecov Report

Merging #1418 into master will decrease coverage by 0.01%.
The diff coverage is 100%.

…refect into scheduled-start-time

roveo · 2019-08-30T11:12:08Z

I think there's something wrong with this. Here's a very simple flow logging it's context:

import prefect
import time
from datetime import timedelta
from pprint import pprint


class Task1(prefect.Task):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def run(self):
        run = prefect.context["scheduled_start_time"].strftime("%Y-%m-%d %H:%M:%S")
        self.logger.warning(pprint(prefect.context.__dict__))
        return 5


@prefect.task
def task2(x):
    pass


schedule = prefect.schedules.CronSchedule("* * * * *")
flow = prefect.Flow("test_flow_name", schedule=schedule)

with flow:
    task2(Task1())


if __name__ == "__main__":
    flow.run()

And here's the output of the first two runs. scheduled_start_time doesn't change between runs.

 ❯ python test_scheduled_start_time.py
[2019-08-30 11:07:17,364] INFO - prefect.Flow: test_flow_name | Waiting for next scheduled run at 2019-08-30T11:08:00+00:00
[2019-08-30 11:08:00,001] INFO - prefect.FlowRunner | Beginning Flow run for 'test_flow_name'
[2019-08-30 11:08:00,002] INFO - prefect.FlowRunner | Starting flow run.
[2019-08-30 11:08:00,014] INFO - prefect.TaskRunner | Task 'Task1': Starting task run...
{'caches': {},
 'checkpointing': False,
 'date': DateTime(2019, 8, 30, 11, 8, 0, 2173, tzinfo=Timezone('UTC')),
 'flow_name': 'test_flow_name',
 'logger': <Logger prefect.Task: Task1 (INFO)>,
 'map_index': None,
 'scheduled_start_time': DateTime(2019, 8, 30, 11, 8, 0, tzinfo=Timezone('UTC')),
 'task_full_name': 'Task1',
 'task_name': 'Task1',
 'task_run_count': 1,
 'task_slug': 'a6bf86d3-2c63-4ad7-b591-0a9c9bd93eeb',
 'task_tags': set(),
 'today': '2019-08-30',
 'today_nodash': '20190830',
 'tomorrow': '2019-08-31',
 'tomorrow_nodash': '20190831',
 'yesterday': '2019-08-29',
 'yesterday_nodash': '20190829'}
[2019-08-30 11:08:00,015] WARNING - prefect.Task: Task1 | None
[2019-08-30 11:08:00,016] INFO - prefect.TaskRunner | Task 'Task1': finished task run for task with final state: 'Success'
[2019-08-30 11:08:00,017] INFO - prefect.TaskRunner | Task 'task2': Starting task run...
[2019-08-30 11:08:00,017] INFO - prefect.TaskRunner | Task 'task2': finished task run for task with final state: 'Success'
[2019-08-30 11:08:00,020] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
[2019-08-30 11:08:00,028] INFO - prefect.Flow: test_flow_name | Waiting for next scheduled run at 2019-08-30T11:09:00+00:00
[2019-08-30 11:09:00,003] INFO - prefect.FlowRunner | Beginning Flow run for 'test_flow_name'
[2019-08-30 11:09:00,003] INFO - prefect.FlowRunner | Starting flow run.
[2019-08-30 11:09:00,015] INFO - prefect.TaskRunner | Task 'Task1': Starting task run...
{'caches': {'Task1': [], 'task2': []},
 'checkpointing': False,
 'date': DateTime(2019, 8, 30, 11, 9, 0, 3252, tzinfo=Timezone('UTC')),
 'flow_name': 'test_flow_name',
 'logger': <Logger prefect.Task: Task1 (INFO)>,
 'map_index': None,
 'scheduled_start_time': DateTime(2019, 8, 30, 11, 8, 0, tzinfo=Timezone('UTC')),
 'task_full_name': 'Task1',
 'task_name': 'Task1',
 'task_run_count': 1,
 'task_slug': 'a6bf86d3-2c63-4ad7-b591-0a9c9bd93eeb',
 'task_tags': set(),
 'today': '2019-08-30',
 'today_nodash': '20190830',
 'tomorrow': '2019-08-31',
 'tomorrow_nodash': '20190831',
 'yesterday': '2019-08-29',
 'yesterday_nodash': '20190829'}
[2019-08-30 11:09:00,023] WARNING - prefect.Task: Task1 | None
[2019-08-30 11:09:00,024] INFO - prefect.TaskRunner | Task 'Task1': finished task run for task with final state: 'Success'
[2019-08-30 11:09:00,024] INFO - prefect.TaskRunner | Task 'task2': Starting task run...
[2019-08-30 11:09:00,025] INFO - prefect.TaskRunner | Task 'task2': finished task run for task with final state: 'Success'
[2019-08-30 11:09:00,025] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded

roveo · 2019-08-30T11:29:04Z

The version from PyPi works correctly:

❯ python test_scheduled_start_time.py
[2019-08-30 11:26:41,345] INFO - prefect.Flow | Waiting for next scheduled run at 2019-08-30T11:27:00+00:00
[2019-08-30 11:27:00,001] INFO - prefect.FlowRunner | Beginning Flow run for 'test_flow_name'
[2019-08-30 11:27:00,004] INFO - prefect.FlowRunner | Starting flow run.
[2019-08-30 11:27:00,013] INFO - prefect.TaskRunner | Task 'Task1': Starting task run...
{'caches': {},
 'checkpointing': False,
 'date': DateTime(2019, 8, 30, 11, 27, 0, 1302, tzinfo=Timezone('UTC')),
 'flow_name': 'test_flow_name',
 'logger': <Logger prefect.Task (INFO)>,
 'map_index': None,
 'scheduled_start_time': DateTime(2019, 8, 30, 11, 27, 0, 1262, tzinfo=Timezone('UTC')),
 'task_full_name': 'Task1',
 'task_name': 'Task1',
 'task_run_count': 1,
 'task_slug': 'c59d838e-3c75-4a38-a36f-fb01a6a70b40',
 'task_tags': set(),
 'today': '2019-08-30',
 'today_nodash': '20190830',
 'tomorrow': '2019-08-31',
 'tomorrow_nodash': '20190831',
 'yesterday': '2019-08-29',
 'yesterday_nodash': '20190829'}
[2019-08-30 11:27:00,015] WARNING - prefect.Task | None
[2019-08-30 11:27:00,015] INFO - prefect.TaskRunner | Task 'Task1': finished task run for task with final state: 'Success'
[2019-08-30 11:27:00,016] INFO - prefect.TaskRunner | Task 'task2': Starting task run...
[2019-08-30 11:27:00,017] INFO - prefect.TaskRunner | Task 'task2': finished task run for task with final state: 'Success'
[2019-08-30 11:27:00,021] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
[2019-08-30 11:27:00,022] INFO - prefect.Flow | Waiting for next scheduled run at 2019-08-30T11:28:00+00:00
[2019-08-30 11:28:00,002] INFO - prefect.FlowRunner | Beginning Flow run for 'test_flow_name'
[2019-08-30 11:28:00,005] INFO - prefect.FlowRunner | Starting flow run.
[2019-08-30 11:28:00,013] INFO - prefect.TaskRunner | Task 'Task1': Starting task run...
{'caches': {'Task1': [], 'task2': []},
 'checkpointing': False,
 'date': DateTime(2019, 8, 30, 11, 28, 0, 3289, tzinfo=Timezone('UTC')),
 'flow_name': 'test_flow_name',
 'logger': <Logger prefect.Task (INFO)>,
 'map_index': None,
 'scheduled_start_time': DateTime(2019, 8, 30, 11, 28, 0, 3228, tzinfo=Timezone('UTC')),
 'task_full_name': 'Task1',
 'task_name': 'Task1',
 'task_run_count': 1,
 'task_slug': 'c59d838e-3c75-4a38-a36f-fb01a6a70b40',
 'task_tags': set(),
 'today': '2019-08-30',
 'today_nodash': '20190830',
 'tomorrow': '2019-08-31',
 'tomorrow_nodash': '20190831',
 'yesterday': '2019-08-29',
 'yesterday_nodash': '20190829'}
[2019-08-30 11:28:00,016] WARNING - prefect.Task | None
[2019-08-30 11:28:00,017] INFO - prefect.TaskRunner | Task 'Task1': finished task run for task with final state: 'Success'
[2019-08-30 11:28:00,017] INFO - prefect.TaskRunner | Task 'task2': Starting task run...
[2019-08-30 11:28:00,018] INFO - prefect.TaskRunner | Task 'task2': finished task run for task with final state: 'Success'
[2019-08-30 11:28:00,018] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded```

jlowin · 2019-08-30T11:35:01Z

Thanks @roveo, looking into why that's happening now!

Allow `temporary_settings` to restore values to their default value

jlowin added 2 commits August 28, 2019 08:58

Handle scheduled start time in local runs (plus tests)

0ab74d2

Add note about concurrent run support

05cd15a

jlowin requested review from cicdw and joshmeek as code owners August 28, 2019 14:59

jlowin added 2 commits August 28, 2019 10:59

Fix typo

f888160

Update CHANGELOG.md

a9895a8

cicdw reviewed Aug 28, 2019

View reviewed changes

docs/guide/core_concepts/flows.md Show resolved Hide resolved

src/prefect/core/flow.py Outdated Show resolved Hide resolved

jlowin added 3 commits August 28, 2019 20:48

Adjust scheduled start time to respect passed values

e2d9789

Format FAQ

bec2fb8

Merge branch 'master' into scheduled-start-time

efc457b

jlowin added 2 commits August 28, 2019 20:58

Hello darkness my old friend

317c8f2

Merge branch 'scheduled-start-time' of https://github.com/PrefectHQ/p…

c2618ef

…refect into scheduled-start-time

cicdw approved these changes Aug 29, 2019

View reviewed changes

Merge branch 'master' into scheduled-start-time

d889e88

cicdw merged commit 71e053d into master Aug 29, 2019

cicdw deleted the scheduled-start-time branch August 29, 2019 04:30

jlowin mentioned this pull request Aug 30, 2019

Scheduled start time not updating between runs #1428

Closed

jlowin mentioned this pull request Aug 30, 2019

Ensure the flow.run() scheduled_start_time updates with each loop #1429

Merged

3 tasks

zanieb added a commit that referenced this pull request Apr 13, 2022

Merge pull request #1418 from PrefectHQ/fixup-remove-vars

ca63cde

Allow `temporary_settings` to restore values to their default value

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Handle scheduled start time in local runs (plus tests) #1418

Handle scheduled start time in local runs (plus tests) #1418

jlowin commented Aug 28, 2019

codecov bot commented Aug 29, 2019 •

edited

Loading

roveo commented Aug 30, 2019

roveo commented Aug 30, 2019

jlowin commented Aug 30, 2019

Handle scheduled start time in local runs (plus tests) #1418

Handle scheduled start time in local runs (plus tests) #1418

Conversation

jlowin commented Aug 28, 2019

What does this PR change?

codecov bot commented Aug 29, 2019 • edited Loading

Codecov Report

roveo commented Aug 30, 2019

roveo commented Aug 30, 2019

jlowin commented Aug 30, 2019

codecov bot commented Aug 29, 2019 •

edited

Loading