https://prefect.io logo
d

Dotan Asselmann

12/10/2020, 7:37 AM
Hey, is it possible to use context value as a default for Parameter?
m

Marwan Sarieddine

12/10/2020, 1:27 PM
I don't think so but perhaps you can set null as the default and then try to fetch the parameter value, if it is None then make use of prefect.context i.e.
Copy code
import random

from prefect import task, Parameter, Flow


@task(name="Task A")
def task_a(x):
    x = x or prefect.context.get("x")
    return x + 2


with Flow("test") as flow:
    x = Parameter("x", default=None)
    task_a(x)
z

Zanie

12/10/2020, 3:38 PM
Thanks for responding Marwan! I think thatÔÇÖs likely the best way to do this
­čĹŹ 1
d

Dotan Asselmann

12/11/2020, 6:38 PM
Thank you. This is great ­čÖé
­čĹŹ 1
p

Pedro Machado

12/12/2020, 12:31 AM
I am testing an approach subclassing the
Parameter
class. I added another keyword argument that receives a callable that provides the default value if the parameter is null. I was using the approach above before but since it's so common in our flows, I wanted to eliminate the extra task. The advantage of not having a separate task is that the Parameter value can be used in templating without having to pass it as an explicit dependency to every task that needs it. You might argue that it's better to have an explicit data dependency, but in my use case the parameter is a date that is used as part of the path of the stored results.
m

Matt Drago

12/13/2020, 10:01 PM
Snap @Pedro Machado, I just implemented
ContextDefaultParameter
and
ContextDefaultDateTimeParameter
last night to solve this issue myself. I was planning on adding it to the Prefect Task Library at some point. I need to add the error handling code as I just wanted to see if I could get it to work.
p

Pedro Machado

12/13/2020, 10:12 PM
Cool, Matt. How did you implement the interface?
m

Matt Drago

12/13/2020, 10:12 PM
Copy code
class ContextDefaultParameter(Parameter):

    def __init__(
        self,
        name: str,
        context_name: str,
        tags: Iterable[str] = None,
    ) -> None:
        super().__init__(name=name, required=False, tags=tags)
        self.context_name = context_name

    def run(self) -> Any:
        value = super().run()

        if value is None:
            value = prefect.context[self.context_name]

        return value
Note, I'm note a native python coder, there was much copy paste change going on here.
One thing I'd want to add to this is raising an error if the context does not have the named field.
p

Pedro Machado

12/13/2020, 10:14 PM
Since you are accessing context with [] instead of
get
, I should already raise an exception if the key is missing.
­čĹŹ 1
m

Matt Drago

12/13/2020, 10:16 PM
Oh yeah, I forgot about the
get
function. Even using
get
though, i think it should raise an error if the context key could not be found.
p

Pedro Machado

12/13/2020, 10:21 PM
Here's what I was playing with. I don't love the name :-)
Copy code
class ParameterWithDynamicDefault(Parameter):
    def __init__(
        self,
        name: str,
        required: bool = True,
        default_callable: Callable[[], JSONSerializableParameterValue] = None,
        tags: Iterable[str] = None,
    ) -> None:

        if default_callable and not callable(default_callable):
            raise ValueError(
                f"default_callable must be a callable. Got {type(default_callable)} instead."
            )

        self.default_callable = default_callable
        required = False if self.default_callable else required
        super().__init__(name, required=required, tags=tags)

    def run(self) -> Any:
        value = super().run()
        if value is None:
            value = self.default_callable()

        return value
m

Matt Drago

12/13/2020, 10:35 PM
Names are tough ­čśë
That's a different angle than I was thinking. The way that I read this functionality is like, "I'll use a value that the flow is called with, but if it's not supplied, I'll run this function to calculate a value".
p

Pedro Machado

12/14/2020, 12:51 AM
Correct. I am using the scheduled start time from the context to compute a relative start date if one is not provided.
m

Matt Drago

12/14/2020, 12:53 AM
I'm doing a similar use case, use the scheduled_start_time if the date to use is not supplied.
This is how I use it
extract_date = ContextDefaultDateTimeParameter(name="extract_date", context_name="scheduled_start_time")
p

Pedro Machado

12/14/2020, 1:01 AM
Got it. I have to use some complex date offset logic and I wanted that encapsulated it in the
Parameter
so I could use it in
target
or
location
templating.
m

Matt Drago

12/14/2020, 1:02 AM
Nice, I can see use cases for both approaches.
3 Views