Ok with the dynamic, I am be able to execute the workflow, but it still show out this error in the task: Traceback (most recent call last): File "/usr/local/lib/python3.10/site-packages/flytekit/exceptions/scopes.py", line 242, in user_entry_point return wrapped(*args, **kwargs) File "/root/workflows/simple_example.py", line 22, in wf installed_pandas_version=check_pandas_version(pandas_version=pandas_version).with_overrides(container_image=image_str) File "/usr/local/lib/python3.10/site-packages/flytekit/core/promise.py", line 486, in with_overrides self.ref.node.with_overrides(*args, **kwargs) File "/usr/local/lib/python3.10/site-packages/flytekit/core/node.py", line 196, in with_overrides assert_not_promise(v, "container_image") File "/usr/local/lib/python3.10/site-packages/flytekit/core/node.py", line 25, in assert_not_promise raise AssertionError(f"Cannot use a promise in the {location} Value: {v}") Message: AssertionError: Cannot use a promise in the container_image Value: Promise(node:dn0.o0.[]) User error.

can you paste your workflow and task definitions here, now that you have updated to use

dynamic

Sure! Thanks! @proud-answer-87162

from flytekit import task, dynamic, workflow
from typing import Dict



@task
def get_image(pandas_version: str) -> str:
    return f"repo/pandas:{pandas_version}"


@task
def check_pandas_version(image: str) -> Dict[str, str]:
    import pandas as pd
    return {"image": image, "installed_pandas_version": pd.__version__}



@dynamic
def wf(pandas_version: str):
    image_str = get_image(pandas_version=pandas_version)
    check_pandas_version(image=image_str).with_overrides(container_image=image_str)


if __name__ == "__main__":
    # Execute the workflow by invoking it like a function and passing in
    # the necessary parameters
    print(f"Running wf() {wf(pandas_version='2.1.0')}")

ok, i'm not super familiar with overrides or

dynamic

workflows, but i think i found a workaround. root cause: as the error stated, you can't use a

promise

for most types of

overrides

. there are some exceptions, but many need to be resolved values, including

container_image

. workaround: call the

get_image

task and (what you call)

wf

from a parent workflow. this will force the result of

get_image

to get resolved before being processed by the

dynamic

wf. I updated the example to use python version instead of pandas because I don't have pandas repo setup like yours. e.g.:

from flytekit import task, dynamic, workflow


@task
def get_image(python_version: str) -> str:
    return f"<http://ghcr.io/flyteorg/flytekit:py{python_version}-latest|ghcr.io/flyteorg/flytekit:py{python_version}-latest>"


@task
def check_python_version(image: str) -> str:
    return f"image: {image}"


@dynamic
def use_dynamic_image(imagename: str) -> str:
    return check_python_version(image=imagename).with_overrides(container_image=imagename)


@workflow
def dynamic_image_wf(python_version: str) -> str:
    imagename = get_image(python_version=python_version)
    return use_dynamic_image(imagename=imagename)


if __name__ == "__main__":
    print(f"Running wf() {dynamic_image_wf(python_version='3.11')}")

Thanks for the response, it looks promising. I will try it and let you know how it works.