> 387 callargs = dict(signature.bind(*args, **kwargs).arguments) # type: Dict 388 389 # bind() compresses all variable keyword arguments under the ** argument name, 7/site-packages/prefect/core/task.py in bind(self, mapped, upstream_tasks, flow, *args, **kwargs)ģ85 # this will raise an error if callargs weren't all provided 386 signature = inspect.signature(n) > 347 *args, mapped=mapped, upstream_tasks=upstream_tasks, flow=flow, **kwargs 7/site-packages/prefect/core/task.py in _call_(self, mapped, task_args, upstream_tasks, flow, *args, **kwargs) > 2 result = foo( 42, skip_on_upstream_skip=False) TypeError Traceback (most recent call last) : result = foo( 42, skip_on_upstream_skip= False) Hey there, is it not possible to set skip_on_upstream_skip when instantiating a task? In : from prefect import task, Flow You might experience some growing pains in setting the correct keys of the task_states dictionary (you'll have to understand when and why task copies are created when building your Flow), but otherwise this should do the trick This way, in future runs, the CachedState which contains the prior result will be re-usedģ.) you'll probably want a way of persisting this data across sessions I'd recommend simply using cloudpickle to dump the CachedState objects to bytes, but if you're feeling brave you could also dig into result handlers and how rialize() works So the first way that comes to my mind is:ġ.) set your tasks to cache their outputs using two simple keywords cache_for and cache_validator ( )Ģ.) use the task_states keyword argument of n to provide a dictionary of type Dict, where each key is a cached task and the corresponding value is the CachedState object associated to a prior run of the flow (extracted from n().result). Hi Great and interesting question there are definitely ways you can implement this yourself using Prefect's current functionality, but it isn't exposed as a first-class operation right now.
0 kommentar(er)
