Rob Fowler
09/08/2023, 9:47 AM@flow
For example, all my flows have the following:
@flow(name='gen-pblock-v2')
def gen_pblock(defines: Dict = None, oformat: str = 'json'):
defines = default_defines(defines)
late_check(defines)
ie, they all take this defines dictionary that gets is sent to default_defines to fill in defaults if not present.
All my flows inspect the context to get the flow schedule start and using defines, maybe don't run if it is too late. Ie the prefect server has been down etc.
As I do this in all flows, can I add a higher decorator that does these things and passes the defines on for further use?redsquare
09/08/2023, 9:54 AMRob Fowler
09/08/2023, 11:25 AM@flow
has no run context, After it does. Here is a transparent wrapper as an example in case anyone else wants the same:
from functools import wraps
def print_args_decorator_with_prefix(prefix=""):
def actual_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
logger = get_run_logger()
<http://logger.info|logger.info>(f"{prefix}Arguments: {args}")
<http://logger.info|logger.info>(f"{prefix}Keyword arguments: {kwargs}")
return func(*args, **kwargs)
return wrapper
return actual_decorator
@flow(name='list-components-v2')
@print_args_decorator_with_prefix(prefix="list")
def list_components(defines: Dict = None):
logger = get_run_logger()
defines = default_defines(defines)
late_check(defines)
assert_defines(defines, ['client', 'config_file'])