https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Can you convert this prefect2 code to...
# ask-marvin
b
@Marvin Can you convert this prefect2 code to prefect3: import prefect from prefect import task, flow, get_run_logger from prefect.deployments import Deployment from prefect.client.schemas.schedules import CronSchedule import knime import os FLOW_NAME: str = 'Centra 08 Stock' DEPLOYMENT: str = 'Retail' KNIME_BASE = 'F:/Analytics/Workflows_knime/NN07/EDI/Madden Analytics/' KNIME_FLOWS = ['Centa 08 Stock Update'] @task(retries=2, retry_delay_seconds=15, task_run_name="{name}") def run_knime(workflow: str, name: str): with knime.Workflow(workflow) as wf: wf.execute() @flow(name=FLOW_NAME) def knime_workflows(): for f in KNIME_FLOWS: knime_flow = os.path.join(KNIME_BASE, f) logger = get_run_logger() logger.info(f'Running {knime_flow}') run_knime(knime_flow, name=f) deployment = Deployment.build_from_flow( flow=knime_workflows, name=DEPLOYMENT, skip_upload=True, schedules = [CronSchedule(cron="40 2 * * 1-6", timezone="Europe/Copenhagen")]) if name == "__main__": deployment.apply()
m
✅ thought for 205.3 seconds
Here’s your Prefect 2 → 3 conversion. Key changes: - Replace Deployment.build_from_flow with flow.deploy - Use prefect.schedules.Cron for timezone-aware cron - Specify a work pool (3.x requires it); for local execution use a process-type pool - No skip_upload in 3.x; use build=False for local code Converted code
Copy code
import os
import knime

from prefect import task, flow, get_run_logger
from prefect.schedules import Cron

FLOW_NAME: str = "Centra 08 Stock"
DEPLOYMENT: str = "Retail"
WORK_POOL_NAME: str = "local"  # change to your work pool name

KNIME_BASE = "F:/Analytics/Workflows_knime/NN07/EDI/Madden Analytics/"
KNIME_FLOWS = ["Centa 08 Stock Update"]

@task(retries=2, retry_delay_seconds=15, task_run_name="{name}")
def run_knime(workflow: str, name: str):
    with knime.Workflow(workflow) as wf:
        wf.execute()

@flow(name=FLOW_NAME)
def knime_workflows():
    logger = get_run_logger()
    for f in KNIME_FLOWS:
        knime_flow = os.path.join(KNIME_BASE, f)
        <http://logger.info|logger.info>(f"Running {knime_flow}")
        run_knime(knime_flow, name=f)

if __name__ == "__main__":
    # Prefect 3.x deployment (replaces Deployment.build_from_flow(...).apply())
    knime_workflows.deploy(
        name=DEPLOYMENT,
        work_pool_name=WORK_POOL_NAME,  # must exist
        schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
        build=False,  # do not build/push an image; run local code
    )
How to run it 1) Create a local process work pool (once)
Copy code
prefect work-pool create "local" --type process
2) Start a worker for that pool (keep this running)
Copy code
prefect worker start --pool "local"
3) Register the deployment (runs the script)
Copy code
python your_script.py
Notes - In Prefect 3.x, 
Deployment.build_from_flow
and 
prefect deployment build
were removed. Use 
flow.deploy(...)
. - 
CronSchedule
moved to a new API—use 
prefect.schedules.Cron(...)
. Passing 
cron="..."
is supported, but use 
Cron(...)
if you need a timezone. - Your 
@task
options (retries, retry_delay_seconds, task_run_name) and 
get_run_logger()
work the same in 3.x. Optional: local scheduler without a worker If you prefer to run on a local scheduler (no worker), swap the deploy call for serve:
Copy code
if __name__ == "__main__":
    knime_workflows.serve(
        name=DEPLOYMENT,
        schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
    )
If you want me to wire this to a Docker/Kubernetes work pool instead, tell me your target environment and I’ll adjust the code and commands.
b
@Marvin My work pool is called default but I get this error when runnig: ValueError: Either an image or remote storage location must be provided when deploying a deployment.
m
✅ thought for 36.5 seconds
That error means your “default” work pool is not a process-type pool (it’s likely Docker/Kubernetes/ECS). Those pool types need either: - an image to run your code, or - a remote source/storage for your code Since your flow calls KNIME on a Windows path (F:/…), the simplest fix is to run it on the same machine via a process work pool. Do this: 1) Inspect your pool’s type
Copy code
prefect work-pool inspect "default"
2) If it’s NOT type “process”, create a process pool and use it
Copy code
prefect work-pool create "windows-process" --type process
prefect worker start --pool "windows-process"
Then deploy with no image (local code), keeping your schedule:
Copy code
knime_workflows.deploy(
    name=DEPLOYMENT,
    work_pool_name="windows-process",
    schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
    build=False,
)
Alternative if you must use the existing “default” pool (Docker/K8s/ECS): - Provide an image (must include Python deps + KNIME if needed) and let Prefect build/push it:
Copy code
knime_workflows.deploy(
    name=DEPLOYMENT,
    work_pool_name="default",
    image="<http://ghcr.io/your-org/knime-runner:latest|ghcr.io/your-org/knime-runner:latest>",
    build=True,
    push=True,
    schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
)
- Or provide a remote code location (e.g., Git repo) so the worker can fetch code:
Copy code
flow.from_source(
    source="<https://github.com/owner/repo.git>",
    entrypoint="path/to/script.py:knime_workflows",
).deploy(
    name=DEPLOYMENT,
    work_pool_name="default",
    schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
    # still ensure the worker image/environment has your dependencies
)
Given your Windows KNIME path, I strongly recommend the process pool route on that machine. If you share the output of:
Copy code
prefect work-pool inspect "default"
I can tailor the exact deploy snippet for your setup.
b
@Marvin the output is this. I'm running a local python file against a remote OSS server: WorkPool( id='04a6f0e9-5268-41cf-8a2a-c944bVVVVV', created=DateTime(2025, 7, 14, 7, 24, 18, 899820, tzinfo=Timezone('UTC')), updated=DateTime(2025, 12, 29, 10, 8, 22, 549493, tzinfo=Timezone('UTC')), name='default', type='process', base_job_template={ 'variables': { 'type': 'object', 'properties': { 'env': {'type': 'object', 'title': 'Environment Variables', 'description': 'Environment variables to set when starting a flow run.', 'additionalProperties': {'anyOf': [{'type': 'string'}, {'type': 'null'}]}}, 'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}], 'title': 'Name', 'default': None, 'description': 'Name given to infrastructure created by a worker.'}, 'labels': {'type': 'object', 'title': 'Labels', 'description': 'Labels applied to infrastructure created by a worker.', 'additionalProperties': {'type': 'string'}}, 'command': { 'anyOf': [{'type': 'string'}, {'type': 'null'}], 'title': 'Command', 'default': None, 'description': 'The command to use when starting a flow run. In most cases, this should be left blank and the command will be automatically generated by the worker.' }, 'working_dir': { 'anyOf': [{'type': 'string', 'format': 'path'}, {'type': 'null'}], 'title': 'Working Directory', 'default': None, 'description': 'If provided, workers will open flow run processes within the specified path as the working directory. Otherwise, a temporary directory will be created.' }, 'stream_output': {'type': 'boolean', 'title': 'Stream Output', 'default': True, 'description': 'If enabled, workers will stream output from flow run processes to local standard output.'} } }, 'job_configuration': {'env': '{{ env }}', 'name': '{{ name }}', 'labels': '{{ labels }}', 'command': '{{ command }}', 'working_dir': '{{ working_dir }}', 'stream_output': '{{ stream_output }}'} }, status=WorkPoolStatus.READY, storage_configuration=WorkPoolStorageConfiguration(), default_queue_id='0e12242e-747f-4c00-8380-VVVVVVV'
m
✅ thought for 181.6 seconds
Thanks for the details — that explains it. What’s happening - You’re deploying to a remote OSS server with a process work pool. - With flow.deploy, Prefect 3.x must know how the worker will fetch your code at run time. - Since you passed an in‑memory flow (local file) and no image/storage/module-path, Prefect raises: “Either an image or remote storage location must be provided…” Fix it (pick one) Option A — Use a remote source (recommended) Point the deployment at a Git repo (or other remote storage) so the worker can pull code.
Copy code
from prefect import flow
from prefect.schedules import Cron

if __name__ == "__main__":
    flow.from_source(
        source="<https://github.com/owner/repo.git>",
        entrypoint="path/to/your_script.py:knime_workflows",
    ).deploy(
        name="Retail",
        work_pool_name="default",
        schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
    )
Then run your worker:
Copy code
prefect worker start --pool "default"
Option B — Use a local path as the source (works if the worker runs on the same machine and can see that path)
Copy code
from pathlib import Path
from prefect import flow
from prefect.schedules import Cron

if __name__ == "__main__":
    flow.from_source(
        source=str(Path(__file__).parent),  # folder containing this script (accessible to the worker)
        entrypoint=f"{Path(__file__).name}:knime_workflows",
    ).deploy(
        name="Retail",
        work_pool_name="default",
        schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
    )
Note: The worker must be on that same machine and have the same filesystem path. Option C — Use a module path (requires your code be importable on the worker) Install your project as a package on the worker (pip install -e .), then:
Copy code
from prefect.types.entrypoint import EntrypointType
from prefect.schedules import Cron

knime_workflows.deploy(
    name="Retail",
    work_pool_name="default",
    schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
    entrypoint_type=EntrypointType.MODULE_PATH,  # worker will import your module
)
Optional: local-only scheduling If you’d rather not involve the server for scheduling (runs on your machine only), use serve:
Copy code
knime_workflows.serve(
    name="Retail",
    schedule=Cron("40 2 * * 1-6", timezone="Europe/Copenhagen"),
)
Docs: Run flows in local processes Given your KNIME path is on Windows, Option A (Git) or Option B (local path the worker can access) are the most straightforward. Which approach would you like to use? I can tailor the exact snippet to your repo/path.
5 Views
Open in Slack
PreviousNext
Powered by