Hi - I had a quick question regarding parameters. Is it possible for a parameter to be dynamic? I have a function within my flow that returns the current date minus 7 days, that is then used as a parameter called process_date. This parameter is set within my flow as such:

with *Flow*("dummy_flow") as flow:

    

process_date = *Parameter*(name="process_date", default=*date_minus_seven*())

The issue with this is the process_date parameter is set at flow registration, but isn’t updated each time the flow runs. For example, if I registered the flow today, the process_date parameter would be set as 2021-07-15, but when the flow runs tomorrow, the process_date would still be set to 2021-07-15.

Hey @Ben Sack, the issue here is the function is executed during build time. The first way to fix this would be moving to script-based storage like Github or S3. The default is a pickle-based storage that evaluates that date and serializes it during build time. This will be true even if you did

default=datetime.datetime.now()

The recommended approach is to have a Parameter with a value like “today”, and then calculate that in a task. This defers the execution during run time.

Thanks! I’ll give it a shot, appreciate the response

Might be helpful

I make heavy use of function wrappers for time-based operations, with the default parameter being None so the intermediate wrapper can detect that and interpolate the correct time value during runtime. There's some extra work to do with inspect and function signatures, though, to get the system to pick everything up correctly

Thanks Sam, interesting approach.. would you happen to have any snippet/example you could provide. We currently use Docker storage for our flows but I’m still curious on how you tackled the issue

Sure, he's a brief snippet that should work with any storage mechanism, where resolve_time is whatever method you use to set your time parameter. If you pay attention, argspec_factory is nearly a concept from the wrapt library that I can't use because wrapt doesn't pickle at all.

import functools

import inspect

from prefect import task, Flow

def argspec_factory(wrapped):

sig = inspect.signature(wrapped)

new_params = []

for name, parameter in sig.parameters.items():

# Perform any filter, updates, etc. as needed

# to function parameters here. If you use annotation

# they need to be converted to strings here.

...

new_params.append(parameter)

sig = sig.replace(parameters=new_params)

def wrap(func):

@functools.wraps(func)

def wrapper(*args, **kwargs):

kwargs['mytime'] = None if kwargs.get('mytime', None) is None else resolve_time(mytime)

return func(*args, **kwargs)

wrapper.__signature__ = argspec_factory(func)

return wrapper

@task

@wrap

def task_func(mytime=None):

# do work

...

with Flow("name"):

# Create flow as normal

That’s clever! Did you mean

@wrap

instead of

@wrapper

around the function?

Nice! Thanks!

@Kevin Kho Yup, I sure did. Fixed