Alon Barad
05/07/2023, 3:00 PMNate
05/07/2023, 11:29 PMcreate a base flow class that can execute tasks based on configurationit sounds like maybe you want to conditionally execute tasks in a flow based on some logic related to configuration, in which case I would think you should be able to do this with a standard
@flow
decorated function, but perhaps I'm not understanding your use case.Alon Barad
05/08/2023, 7:21 AMclass BaseFlow:
def __init__(self, actions, config):
self.actions = actions
self.config = config
@task(name="Crawl")
def crawl(self, *args, **kwargs):
raise NotImplementedError
@task(name="Parse")
def parse(self, *args, **kwargs):
raise NotImplementedError
@flow(name="Base Flow")
def run(self):
if 'crawl' in self.actions:
self.crawl(**self.actions['crawl'])
if 'parse' in self.actions:
self.parse(**self.actions['parse'])
I want to be able to run every task \ the whole flow with a custom config JSON from the UI
Thanks!Miguel Moncada
05/08/2023, 7:28 AM@flow(name="parent_flow")
def parent_flow(env: str = "staging")
# Logic to fetch runtime values depending on param from secret/variables
...
Apologies if this is not what you were looking for šAlon Barad
05/08/2023, 7:33 AMMiguel Moncada
05/08/2023, 7:42 AM.
āāā __init__.py
āāā _version.py
āāā deployments
ā āāā __init__.py
ā āāā first_flow_deployment.py
āāā flows
āāā __init__.py
āāā first_flow.py
āāā second_flow.py
āāā utils
āāā __init__.py
āāā commons
ā āāā __init__.py
ā āāā common_flow.py