Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Make new conditional API compatible with the imperative API #2538

Closed
jlowin opened this issue May 11, 2020 · 3 comments 路 Fixed by #2546
Closed

Make new conditional API compatible with the imperative API #2538

jlowin opened this issue May 11, 2020 · 3 comments 路 Fixed by #2546
Assignees
Labels
enhancement An improvement of an existing feature

Comments

@jlowin
Copy link
Member

jlowin commented May 11, 2020

Current behavior

Please describe how the feature works today
In #2443, @jcrist added a new conditional API to Prefect, which is 馃挴. However, at this moment, it can only be used with the functional API (as it requires a flow in the Prefect context).

Proposed behavior

Please describe your proposed change to the current behavior

The check for case in context currently takes place in Task.bind(), which is called by the functional API but not common imperative patterns like Task.set_upstream(), Task.set_downstream(), Task.set_dependencies(), Flow.add_task(), or Flow.set_dependencies().

I believe Flow.add_task() is the lowest-level API (called by all the others mentioned above) and consequently might be an opportune place to move the check for case in context. This way, adding a task to a flow while in a conditional block would automatically add it to the case class as well. Since the flow is (obviously) available when Flow.add_task() is called, it can be used inside the case context manager when calling bind() and set_upstream().

I think this would enable the following imperative syntax, while still using the existing case context managers:

Switch = Task()
A = Task()
B = Task()
C = Task()
f = Flow('imperative flow')

switch = Switch()
f.add_task(switch)

with case(switch, 'foo'):
    f.add_task(A())

with case(switch, 'bar'):
    b = B()
    c = C()
    c.set_upstream(b, flow=f)

Example

Please give an example of how the enhancement would be useful

@jlowin jlowin added the enhancement An improvement of an existing feature label May 11, 2020
@jcrist
Copy link

jcrist commented May 11, 2020

case needs a reference to the current flow to work properly, so I think this would have to be more like:

with case(switch, 'bar', flow=f):
    ...

@jlowin
Copy link
Member Author

jlowin commented May 11, 2020

I was playing with that, but since Case.add_task() is always called from a parent method that has a flow object available, I think we can actually hide that from the user. For example, in the current implementation, it's called from Task.bind() which has a flow object in scope, so the signature could be Case.add_task(task: Task, flow: Flow=None). We might want to validate that all passed flows are the same for a given case context manager, but by passing it to add_task we can probably hide the need for users to pass it explicitly

@jcrist
Copy link

jcrist commented May 12, 2020

See #2546.

abrookins pushed a commit that referenced this issue Jul 27, 2022
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
enhancement An improvement of an existing feature
Projects
None yet
Development

Successfully merging a pull request may close this issue.

2 participants