Template to schedule KNIME workflows using Prefect

prefect.py

#prefect imports
import prefect
from prefect import task, Flow
from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock
from prefect.storage import Local
import pendulum
import knime
import os

# KNIME FLOW
knime_flow: str = ''

PY_SCRIPT: str = 'full_path_to_this_script.py'
FLOW_NAME: str = 'Your Flow name'
PROJECT: str = 'Operations'

KNIME_BASE = 'Path/to/your/knime/workspace/'
KNIME_FLOWS =  ['Gofact Loader']

@task(max_retries=2, retry_delay=pendulum.duration(seconds=30))
def run_knime(workflow: str):
    with knime.Workflow(workflow) as wf:
        wf.execute()

# Set Cron to run at 5am every day
schedule = Schedule(clocks=[CronClock("0 5 * * *",start_date=pendulum.now())])

with Flow(FLOW_NAME, schedule, storage=Local(path=PY_SCRIPT, stored_as_script=True)) as flow:
    logger = prefect.context.get("logger")
    gofact_process = gofact()
    logger.info('Downloaded GoFact files')
    for f in KNIME_FLOWS:
        knime_flow = os.path.join(KNIME_BASE, f)
        logger.info(f'Running {knime_flow}')
        run_knime(knime_flow, upstream_tasks=[gofact_process])

# Register the flow under the "tutorial" project
flow.register(project_name=PROJECT)