https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Using Prefect 3.6.12, how can I use P...
# ask-marvin
n
@Marvin Using Prefect 3.6.12, how can I use Python to schedule a deployment to run every ten minutes?
m
✅ thought for 79.6 seconds
In Prefect 3.6.12, the simplest way is to set an interval schedule when you create the deployment in Python.
Copy code
python
from datetime import timedelta
from prefect import flow

@flow
def my_flow():
    print("running")

if __name__ == "__main__":
    my_flow.deploy(
        name="every-10-minutes",
        work_pool_name="my-work-pool",
        interval=timedelta(minutes=10),
    )
You can also use seconds:
Copy code
python
interval=600
Or a cron schedule if you prefer cron syntax:
Copy code
python
from prefect import flow

@flow
def my_flow():
    print("running")

if __name__ == "__main__":
    my_flow.deploy(
        name="every-10-minutes",
        work_pool_name="my-work-pool",
        cron="*/10 * * * *",
    )
Recommendation: for “every 10 minutes”, 
interval=timedelta(minutes=10)
is the clearest option. A couple of notes: - 
deploy()
creates/applies the deployment in Prefect 3.x - you’ll need a worker running for the deployment’s work pool - if you want multiple schedules or more control, use the 
schedules=
argument Relevant docs: - Deploy via Python - Create schedules If you want, I can also show the exact version for: - serverless / managed execution - Docker work pools - Kubernetes work pools - or CI/CD (like CodeBuild)
n
@Marvin I don't want to create a new deployment, I want to take an already existing deployment and set it to run every ten minutes
m
✅ thought for 65.5 seconds
Yes — for an existing deployment, you should use the Prefect client to read its schedules and then update one of them. Here’s the Python pattern for Prefect 3.6.12:
Copy code
python
from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def set_deployment_to_every_10_minutes():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        schedules = await client.read_deployment_schedules(deployment.id)

        if not schedules:
            print("This deployment has no schedules to update.")
            return

        schedule = schedules[0]

        await client.update_deployment_schedule(
            deployment_id=deployment.id,
            schedule_id=schedule.id,
            schedule=IntervalSchedule(interval=timedelta(minutes=10)),
            active=True,
        )

        print("Updated deployment schedule to every 10 minutes.")
A few important notes: - 
read_deployment_by_name()
expects 
"<flow-name>/<deployment-name>"
- 
read_deployment_schedules()
returns the schedules attached to that deployment - 
update_deployment_schedule()
updates one existing schedule - if the deployment has multiple schedules, you may want to update all of them, or pick the right one by 
slug
If you want to update all existing schedules on the deployment:
Copy code
python
from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def set_all_deployment_schedules_to_every_10_minutes():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        schedules = await client.read_deployment_schedules(deployment.id)

        for schedule in schedules:
            await client.update_deployment_schedule(
                deployment_id=deployment.id,
                schedule_id=schedule.id,
                schedule=IntervalSchedule(interval=timedelta(minutes=10)),
                active=True,
            )

        print(f"Updated {len(schedules)} schedule(s).")
If the deployment has no schedule yet, then you’d need to create one instead of updating one. If that’s your case, say so and I can give you the exact code for adding a schedule to an existing deployment. Docs: - Deployment client API - Create schedules
n
@Marvin are there other ways to run a deployed flow every ten minutes that don't involve modifying the deployment? I deploy my flows in a CICD pipeline, and this constantly overrides the deployments that already exist, deleting the schedule
m
✅ thought for 112.4 seconds
Yes — if your CI/CD redeploys keep wiping deployment schedules, the best pattern is to schedule runs outside the deployment. The main options are: 1. Use an external scheduler to trigger the deployment every 10 minutes This is usually the simplest and most robust approach. For example, every 10 minutes run:
Copy code
bash
prefect deployment run "my-flow/my-deployment"
That can be done from: - cron - Kubernetes 
CronJob
- GitHub Actions scheduled workflow - AWS EventBridge + Lambda / ECS task / CodeBuild - any other scheduler you already trust In Python, the equivalent is:
Copy code
python
from prefect.deployments import run_deployment

run_deployment(name="my-flow/my-deployment")
Or with the client API:
Copy code
python
from prefect.client.orchestration import get_client

async def trigger():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name("my-flow/my-deployment")
        await client.create_flow_run_from_deployment(deployment.id)
Why this is good for your case: redeploying the deployment does not affect the external scheduler. 2. Use a Prefect Automation Automations are separate from deployment definitions, so they generally survive redeploys better than deployment schedules. An automation can run a deployment as an action. That said, for a pure “every 10 minutes” heartbeat-style schedule, an external scheduler is often simpler than trying to model time-based recurrence through automations. Docs: - Automations - Run deployments 3. Change your CI/CD so it does not overwrite schedules If you want to keep using deployment-native schedules, another option is to make your deployment config the source of truth and include the schedule in CI/CD itself. But from your description, it sounds like you specifically want to avoid that. My recommendation for your setup Since you’re deploying from CI/CD, I’d strongly recommend: - deploy the flow without a schedule - use an external scheduler to call 
prefect deployment run "my-flow/my-deployment"
every 10 minutes That keeps: - deployment definition = managed by CI/CD - execution cadence = managed independently If you want, I can give you a ready-to-use example for: - AWS EventBridge - CodeBuild - GitHub Actions - Kubernetes CronJob - or plain cron
n
@Marvin I want to ADD a new schedule, not update existing ones
m
✅ thought for 57.2 seconds
Got it — if you want to add a new schedule to an existing deployment, use 
create_deployment_schedules(...)
. That appends a schedule; it does not replace existing ones. Example for every 10 minutes:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import IntervalSchedule

async def add_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        created = await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                DeploymentScheduleCreate(
                    schedule=IntervalSchedule(interval=timedelta(minutes=10)),
                    active=True,
                    slug="every-10-minutes",
                )
            ],
        )

        print(created)
If you prefer cron syntax instead:
Copy code
python
from prefect.client import get_client
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import CronSchedule

async def add_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                DeploymentScheduleCreate(
                    schedule=CronSchedule(cron="*/10 * * * *", timezone="UTC"),
                    active=True,
                    slug="every-10-minutes",
                )
            ],
        )
A few notes: - 
slug
is optional, but useful if you want to refer to this schedule later - 
read_deployment_by_name()
expects 
"<flow-name>/<deployment-name>"
- this adds the schedule outside the deployment spec, which can help if your CI/CD deployment process is replacing the deployment definition Docs: - Deployment client API - Create schedules If you want, I can also give you: - a sync version - a version that checks whether the slug already exists first - or a small utility that re-adds the schedule after CI/CD deploys
n
@Marvin It accepts 
list[tuple[SCHEDULE_TYPES, bool]]
for the schedules parameter
m
✅ thought for 73.6 seconds
Yes — you’re right. In Prefect 3.6.12, 
create_deployment_schedules()
accepts either: - 
list[tuple[SCHEDULE_TYPES, bool]]
- 
list[DeploymentScheduleCreate]
So for your version, this is valid and probably the simplest way to add a new schedule to an existing deployment:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def add_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                (
                    IntervalSchedule(interval=timedelta(minutes=10)),
                    True,
                )
            ],
        )
Or with cron:
Copy code
python
from prefect.client import get_client
from prefect.client.schemas.schedules import CronSchedule

async def add_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                (
                    CronSchedule(cron="*/10 * * * *", timezone="UTC"),
                    True,
                )
            ],
        )
A useful nuance: - the tuple form is shorthand for 
(schedule, active)
- if you want per-schedule 
slug
, 
parameters
, or 
max_scheduled_runs
, use 
DeploymentScheduleCreate
- if you pass 
slug=...
or 
parameters=...
to 
create_deployment_schedules(...)
with the tuple form, those values get applied uniformly during conversion So for just add a 10-minute schedule, the tuple form is perfectly fine. Docs: - Create schedules - Deployment client API If you want, I can give you a tiny helper like 
ensure_schedule_exists(...)
so your CI/CD can safely re-add the schedule only if it’s missing.
n
@Marvin This is the function, it does not accepts DeploymentScheduleCreate
Copy code
def create_deployment_schedules(
        self,
        deployment_id: UUID,
        schedules: list[tuple["SCHEDULE_TYPES", bool]],
    ) -> list["DeploymentSchedule"]:
        """
        Create deployment schedules.

        Args:
            deployment_id: the deployment ID
            schedules: a list of tuples containing the schedule to create
                       and whether or not it should be active.

        Raises:
            RequestError: if the schedules were not created for any reason

        Returns:
            the list of schedules created in the backend
        """
        from prefect.client.schemas.actions import DeploymentScheduleCreate
        from prefect.client.schemas.objects import DeploymentSchedule

        deployment_schedule_create = [
            DeploymentScheduleCreate(schedule=schedule[0], active=schedule[1])
            for schedule in schedules
        ]

        json = [
            deployment_schedule_create.model_dump(mode="json")
            for deployment_schedule_create in deployment_schedule_create
        ]
        response = self.request(
            "POST",
            "/deployments/{id}/schedules",
            path_params={"id": deployment_id},
            json=json,
        )
        return DeploymentSchedule.model_validate_list(response.json())
m
✅ thought for 6.3 seconds
You’re absolutely right — thanks for pasting the actual function. Given your installed version, the correct call is only the tuple form: - 
schedules: list[tuple[SCHEDULE_TYPES, bool]]
So to add a 10-minute schedule to an existing deployment in your environment, use:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def add_ten_minute_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                (
                    IntervalSchedule(interval=timedelta(minutes=10)),
                    True,
                )
            ],
        )
Or with cron:
Copy code
python
from prefect.client import get_client
from prefect.client.schemas.schedules import CronSchedule

async def add_ten_minute_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                (
                    CronSchedule(cron="*/10 * * * *", timezone="UTC"),
                    True,
                )
            ],
        )
So the key point is: - you do not pass 
DeploymentScheduleCreate
- Prefect constructs 
DeploymentScheduleCreate
internally from each 
(schedule, active)
tuple Thanks for the correction — I should have matched your exact installed API rather than a broader variant. If helpful, I can also give you: - a sync version - code to list existing schedules first - or an 
ensure_10_min_schedule()
helper that avoids adding duplicates
n
@Marvin What is the slug going to be? I cannot set it if it does not accept a DeploymentScheduleCreate object
m
✅ thought for 7.1 seconds
In your version/API shape, you cannot set the slug when calling 
create_deployment_schedules()
. From the function you pasted, each tuple is converted internally like this:
Copy code
python
DeploymentScheduleCreate(schedule=schedule[0], active=schedule[1])
So only these two fields are populated: - 
schedule
- 
active
That means: - no 
slug
- no custom 
parameters
- no 
max_scheduled_runs
So the answer is: The slug will be auto-generated by the backend, or left unset depending on how that backend version handles schedule creation — but you do not control it through this client method. If you need to know what was created, capture the returned schedules:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def add_schedule():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        created = await client.create_deployment_schedules(
            deployment_id=deployment.id,
            schedules=[
                (
                    IntervalSchedule(interval=timedelta(minutes=10)),
                    True,
                )
            ],
        )

        for s in created:
            print("id:", s.id)
            print("slug:", s.slug)
            print("active:", s.active)
            print("schedule:", s.schedule)
If your real goal is “add this exact schedule only once”, then since you can’t rely on setting a slug, the practical approach is: 1. read the deployment’s existing schedules 2. compare their schedule definitions 3. only create the new one if an equivalent 10-minute schedule is not already present If you want, I can write you a small 
ensure_interval_schedule(deployment_name, minutes=10)
helper for your exact API version.
n
@Marvin What is the point of DeploymentScheduleCreate, then?
m
✅ thought for 8.7 seconds
DeploymentScheduleCreate
is the request schema the client sends to the API. In the code you pasted, the public method accepts a simplified input:
Copy code
python
list[tuple[SCHEDULE_TYPES, bool]]
and then internally converts each tuple into:
Copy code
python
DeploymentScheduleCreate(schedule=..., active=...)
So its purpose is still real — it’s just not exposed as part of that particular method’s public interface in your version. In other words: - you call 
create_deployment_schedules(..., schedules=[(schedule, active)])
- the client builds 
DeploymentScheduleCreate(...)
- the API receives serialized 
DeploymentScheduleCreate
payloads Why have the model at all? 1. It defines the API payload shape It is the canonical schema for “create a deployment schedule”. 2. It validates data before sending The client can ensure the outgoing payload matches what the server expects. 3. It is reused elsewhere in the codebase Even if this specific helper hides it, the schema is still useful internally and in other code paths. 4. It leaves room for richer APIs in other versions Some versions expose richer creation methods; others expose only the tuple shorthand. So in your version, 
DeploymentScheduleCreate
is mostly an internal transport/validation model for this method, not something you pass directly. That’s also why the mismatch feels odd: the underlying schema supports fields like 
slug
, but this convenience method only exposes: - schedule - active So effectively, the wrapper is narrower than the underlying API model. If you want, I can help check whether: - there is another client method in 3.6.12 that exposes 
slug
- or whether the only practical route in your version is using the raw REST API directly.
n
@Marvin Via the UI I can set the slug on an existing deployment when adding a schedule, why can't I do it via the Python SDK?
m
✅ thought for 114.0 seconds
Because this is an SDK limitation in your specific version, not a platform limitation. What’s happening is: - the backend API supports 
slug
- the UI talks to that backend API directly and can send 
slug
- but the Python SDK method in your installed version exposes only:
Copy code
python
  schedules: list[tuple[SCHEDULE_TYPES, bool]]
so it only lets you provide: - the schedule - whether it is active In other words, the UI and API are richer than the convenience wrapper you have in the SDK. So the short answer is: You can do it in the UI because the API supports it; you cannot do it through that Python helper because your SDK version does not expose the field. If you want to do it from Python without upgrading, the workaround is to call the underlying API through the client’s lower-level 
request()
method. Example:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def add_schedule_with_slug():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        payload = [
            {
                "schedule": IntervalSchedule(
                    interval=timedelta(minutes=10)
                ).model_dump(mode="json"),
                "active": True,
                "slug": "every-10-minutes",
            }
        ]

        response = await client.request(
            "POST",
            "/deployments/{id}/schedules",
            path_params={"id": deployment.id},
            json=payload,
        )

        return response.json()
If your client is sync instead of async, the shape is the same, just without 
await
. So your options are: 1. Keep using the SDK helper Easy, but no 
slug
2. Use 
client.request(...)
directly Lets you send 
slug
3. Upgrade to a newer Prefect version Newer versions expanded the schedule client methods to expose more fields Given your CI/CD setup, option 2 is probably the most practical if you need 
slug
now. If you want, I can give you a fully correct helper for your version that: - reads the deployment by name - posts a schedule with 
slug
- returns the created schedule object cleanly
n
@Marvin I cannot even set the schedule parameters via python SDK in Prefect 3.6.12, is this correct?
m
✅ thought for 9.3 seconds
Yes — with the 
create_deployment_schedules()
helper in Prefect 3.6.12, that is correct. From the function you pasted, the SDK only exposes: - 
schedule
- 
active
So when adding a schedule to an existing deployment via that method, you cannot set: - 
slug
- 
parameters
- 
max_scheduled_runs
through the public Python helper. That’s why the UI can do more than that SDK method in your version. So specifically:
Copy code
python
await client.create_deployment_schedules(
    deployment_id=deployment.id,
    schedules=[(my_schedule, True)],
)
does not let you attach schedule-level parameter overrides. If you need schedule parameters in 3.6.12, your choices are: 1. Use the UI Works because it hits the API directly 2. Use the low-level client request You can send the full payload yourself, including 
parameters
Example:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def add_schedule_with_parameters():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

        payload = [
            {
                "schedule": IntervalSchedule(
                    interval=timedelta(minutes=10)
                ).model_dump(mode="json"),
                "active": True,
                "slug": "every-10-minutes",
                "parameters": {
                    "foo": "bar",
                    "x": 123
                },
                "max_scheduled_runs": 100,
            }
        ]

        response = await client.request(
            "POST",
            "/deployments/{id}/schedules",
            path_params={"id": deployment.id},
            json=payload,
        )

        return response.json()
3. Upgrade Prefect Later versions expose more of this through the SDK method itself So the answer to your question is: Yes — in Prefect 3.6.12, the high-level Python SDK method for adding deployment schedules does not let you set schedule parameters. If you want, I can turn the raw 
client.request(...)
approach into a small reusable helper like:
Copy code
python
add_deployment_schedule(
    deployment_name="flow/deployment",
    every_minutes=10,
    slug="every-10-minutes",
    parameters={"foo": "bar"},
)
n
@Marvin this is how you do it
Copy code
with get_client(sync_client=True) as client:
    deployment = client.read_deployment_by_name("fire-sight-polling/fire-sight")

    client.update_deployment(
        deployment.id,
        DeploymentUpdate(
            schedules=[DeploymentScheduleUpdate(schedule=IntervalSchedule(interval=timedelta(minutes=10)), slug="test")]
        ),
    )
m
✅ thought for 9.3 seconds
Yep — that’s the right approach for your version/use case. Thanks for sharing it. So in Prefect 3.6.12, if you want to set schedule fields like 
slug
via Python, using:
Copy code
python
client.update_deployment(
    deployment.id,
    DeploymentUpdate(
        schedules=[
            DeploymentScheduleUpdate(
                schedule=IntervalSchedule(interval=timedelta(minutes=10)),
                slug="test",
            )
        ]
    ),
)
is the way to do it. Your full example:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.actions import DeploymentUpdate
from prefect.client.schemas.schedules import DeploymentScheduleUpdate, IntervalSchedule

with get_client(sync_client=True) as client:
    deployment = client.read_deployment_by_name("fire-sight-polling/fire-sight")

    client.update_deployment(
        deployment.id,
        DeploymentUpdate(
            schedules=[
                DeploymentScheduleUpdate(
                    schedule=IntervalSchedule(interval=timedelta(minutes=10)),
                    slug="test",
                )
            ]
        ),
    )
Important behavior note: - this goes through 
update_deployment(...)
- 
DeploymentUpdate.schedules
lets you provide schedule objects with richer fields - this is different from 
create_deployment_schedules(...)
, which in your installed version is the narrower tuple-based helper One thing to be careful about: 
update_deployment(..., DeploymentUpdate(schedules=[...]))
may replace the deployment’s schedules rather than append one alongside existing schedules. So if your goal is specifically add a new schedule without removing existing ones, the safe pattern is: 1. read the existing deployment 2. take its current schedules 3. append your new 
DeploymentScheduleUpdate(...)
4. send the full combined list back in 
DeploymentUpdate(...)
If you want, I can help write a small 
append_schedule_to_deployment(...)
helper that: - preserves existing schedules - avoids duplicate slugs - works in sync or async mode And thank you for the correction — that’s a useful nuance for 3.6.12.
n
@Marvin
update_deployment(..., DeploymentUpdate(schedules=[...])) may replace the deployment’s schedules rather than append one alongside existing schedules.
it doesn't, I have just tested it
m
✅ thought for 6.9 seconds
Good catch — thanks for verifying. You’re right: in your tested Prefect 3.6.12 behavior, passing 
schedules=[...]
to 
update_deployment(...)
adds the schedule rather than replacing all existing schedules. So the corrected guidance is: - 
create_deployment_schedules(...)
is a narrow helper - 
update_deployment(..., DeploymentUpdate(schedules=[DeploymentScheduleUpdate(...)]))
is the more capable Python SDK path in your version - and in your testing, it does not wipe existing schedules So for adding a 10-minute schedule with a slug, this is a valid pattern:
Copy code
python
from datetime import timedelta
from prefect.client import get_client
from prefect.client.schemas.actions import DeploymentUpdate
from prefect.client.schemas.schedules import (
    DeploymentScheduleUpdate,
    IntervalSchedule,
)

with get_client(sync_client=True) as client:
    deployment = client.read_deployment_by_name("fire-sight-polling/fire-sight")

    client.update_deployment(
        deployment.id,
        DeploymentUpdate(
            schedules=[
                DeploymentScheduleUpdate(
                    schedule=IntervalSchedule(interval=timedelta(minutes=10)),
                    slug="test",
                )
            ]
        ),
    )
And based on what you found, this route also lets you set things the narrower helper does not, such as: - 
slug
- schedule details beyond the tuple shorthand - potentially other schedule-level fields supported by 
DeploymentScheduleUpdate
So the practical lesson here is: In Prefect 3.6.12, for adding richer schedules to an existing deployment from Python, 
update_deployment(... DeploymentUpdate(schedules=[...]))
is the better SDK route than 
create_deployment_schedules(...)
. Thanks again — that’s a genuinely important distinction.
2 Views
Open in Slack
PreviousNext
Powered by