pyflyte run imperative workflows

[pingsutw]

TL;DR

support running imperative workflow by using pyflyte run.
flyteorg/flyte#2706

we try to get the module name from extract_task_module, but it doesn't work for imperative workflow.
You will get the below error when using pyflyte run to run an imperative workflow.

(flytekit-3.9) ➜  flytekit git:(master) ✗ pyflyte --config ~/.flyte/config-remote.yaml run --remote example_imperative.py wf --in1 2 --in2 3
/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/paramiko/transport.py:236: CryptographyDeprecationWarning: Blowfish has been deprecated
  "class": algorithms.Blowfish,
Traceback (most recent call last):
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/bin/pyflyte", line 8, in <module>
    sys.exit(main())
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 1128, in __call__
    return self.main(*args, **kwargs)
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 1053, in main
    rv = self.invoke(ctx)
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 1659, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 1659, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 1659, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 1395, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/kevin/opt/anaconda3/envs/flytekit-3.9/lib/python3.9/site-packages/click/core.py", line 754, in invoke
    return __callback(*args, **kwargs)
  File "/Users/kevin/git/flytekit/flytekit/clis/sdk_in_container/run.py", line 539, in _run
    remote_entity = remote.register_script(
  File "/Users/kevin/git/flytekit/flytekit/remote/remote.py", line 596, in register_script
    upload_location, md5_bytes = fast_register_single_script(
  File "/Users/kevin/git/flytekit/flytekit/tools/script_mode.py", line 102, in fast_register_single_script
    _, mod_name, _, script_full_path = extract_task_module(wf_entity)
  File "/Users/kevin/git/flytekit/flytekit/core/tracker.py", line 237, in extract_task_module
    name = f.__name__.split(".")[-1]
AttributeError: 'ImperativeWorkflow' object has no attribute '__name__'

we can't also use inspect.getmodule to get module name because inspect.getmodule(<imperative_wf>).__name__ is always equal to flytekit.core.workflow

basically, it's unnecessary to get a module name from an entity by calling extract_task_module because we already have a module name that users specify in the pyflyte run command.

To address this issue, we can use the module name passed in the pyflyte run instead

Type

• Bug Fix
• Feature
• Plugin

Are all requirements met?

• Code completed
• Smoke tested
• Unit tests added
• Code documentation added
• Any pending items have an associated Issue

Complete description

How did you fix the bug, make the feature etc. Link to any design docs etc

Tracking Issue

flyteorg/flyte#2706

Follow-up issue

NA

[codecov]

Codecov Report

Merging #1131 (7067721) into master (73eaad1) will increase coverage by 0.12%.
The diff coverage is 80.85%.

@@            Coverage Diff             @@
##           master    #1131      +/-   ##
==========================================
+ Coverage   68.26%   68.38%   +0.12%     
==========================================
  Files         287      288       +1     
  Lines       25829    25963     +134     
  Branches     2885     2899      +14     
==========================================
+ Hits        17633    17756     +123     
- Misses       7719     7728       +9     
- Partials      477      479       +2     

Impacted Files	Coverage Δ
flytekit/remote/remote.py	41.11% <ø> (ø)
flytekit/tools/script_mode.py	49.25% <0.00%> (+0.72%)	⬆️
flytekit/core/tracker.py	48.67% <60.00%> (+0.04%)	⬆️
tests/flytekit/unit/cli/pyflyte/imperative_wf.py	86.36% <86.36%> (ø)
flytekit/clis/sdk_in_container/constants.py	100.00% <100.00%> (ø)
flytekit/clis/sdk_in_container/run.py	83.60% <100.00%> (ø)
tests/flytekit/unit/cli/pyflyte/test_run.py	98.87% <100.00%> (+0.06%)	⬆️
flytekit/extras/tasks/shell.py	77.71% <0.00%> (-0.45%)	⬇️
...ests/flytekit/unit/core/test_structured_dataset.py	98.88% <0.00%> (-0.31%)	⬇️
flytekit/core/data_persistence.py	33.80% <0.00%> (-0.01%)	⬇️
... and 13 more

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

[wild-endeavor]

will take a look tomorrow

[pingsutw]

We can't get the module name of the imperative workflow by using inspect.getmodule because inspect.getmodule(<imperative_wf>).__name__ is always equal to flytekit.core.workflow

To address this issue, we can use the module name passed in the pyflyte run instead.
pyflyte run --remote imperative_wf.py (model_name) wf

[wild-endeavor]

yeah sounds good, but can we make it like pyflyte register and pyflyte package in the future? can we make another issue for that? That will get us around this problem and maybe we can revert this change in the future.

you tested this in a subfolder right? like if you have a/b/c/wf.py running pyflyte run in the c/ folder, still gets you a.b.c.wf as the name.

[pingsutw]

yes, It's working in the subfolder as well. I've tested it.

can we make it like pyflyte register and pyflyte package in the future?

Not sure what you mean. The reason why it only doesn't work in the pyflyte run is that mod_name in the compress_single_script is generated from extract_task_module, but extract_task_module doesn't work for the imperative workflow. Therefore, we have to use the module name passed from the pyflyte run command.

On the other hand, the pyflyte register uses fast_package, which doesn't require module name. That's why the pyflyte register works now.