Hi everyone. I’m trying to dynamically set resources based on the amount of rows in a dataset. I’m hitting this really opaque error that I can’t get past. In trying to get to the bottom of the issue, I adapted the example in the docs to look a lot like my code. I get the same error! Any wisdom would be very much appreciated! Have I formulated this incorrectly?

import typing

from flytekit import Resources, task, workflow, dynamic


@task(requests=Resources(cpu="1", mem="200Mi"), limits=Resources(cpu="2", mem="350Mi"))
def count_unique_numbers_1(x: typing.List[int]) -> int:
    s = set()
    for i in x:
        s.add(i)
    return len(s)


@task
def square_1(x: int) -> int:
    return x * x


@task
def memory_string() -> str:
    return "400Mi"


@workflow
def my_pipeline(x: typing.List[int]) -> int:
    return square_1(x=count_unique_numbers_1(x=x)).with_overrides(
        limits=Resources(cpu="6", mem="500Mi")
    )


@dynamic
def my_workflow_dynamic(x: typing.List[int]) -> int:
    memory = memory_string()
    return square_1(x=count_unique_numbers_1(x=x)).with_overrides(
        limits=Resources(mem=memory)
    )


@workflow
def my_workflow_dynamic_wrapper(x: typing.List[int]) -> int:
    return my_workflow_dynamic(x=x)

And the error is:

TypeError: bad argument type for built-in operation

Hi Joe, Every result during the workflow stage is wrapped as a 'Promise'. To get at the underlying python object, you need to send it to another workflow or task. so in

my_workflow_dynamic

you're actually creating a resources object where memory is defined as a wrapped promise object, which it doesn't want.