Dvir Reingewirtz
07/29/2025, 4:43 PMMarvin
07/29/2025, 4:43 PMMarvin
07/29/2025, 4:43 PMpython
schedules = [
{
"slug": "daily-run", # Add this line
"cron": "0 0 * * *"
},
{
"slug": "weekly-backup", # Add this line
"cron": "0 0 * * 0"
}
]
To fix your error:
1. Make sure each schedule in your configuration has a unique slug
2. The slug should be a simple string identifier (like "daily-run" or "hourly-sync")
3. Use the same slug consistently if you want to update an existing schedule
If you're using YAML for deployment configuration, it would look like this:
yaml
schedules:
- slug: daily-run
cron: "0 0 * * *"
- slug: weekly-backup
cron: "0 0 * * 0"
Would you mind sharing how you're currently setting up your schedules? That way, I can provide more specific guidance on how to add the required slugs to your configuration.Dvir Reingewirtz
07/29/2025, 4:47 PMdef slugify_schedule(sched: Any) -> str:
try:
if hasattr(sched, "model_dump"):
sched_dict = sched.model_dump()
elif hasattr(sched, "dict"):
sched_dict = sched.dict()
elif isinstance(sched, dict):
sched_dict = sched
else:
sched_dict = sched.__dict__
except Exception:
sched_dict = str(sched)
stringified = json.dumps(sched_dict, sort_keys=True, default=str)
hashed = hashlib.sha1(stringified.encode()).hexdigest()[:10]
return f"schedule-{hashed}"
def add_slugs_to_schedules(schedules: list[Any]) -> list[dict]:
updated_schedules = []
for sched in schedules:
slug = slugify_schedule(sched)
updated_schedules.append({
"schedule": sched,
"slug": slug,
})
return updated_schedules
schedules = add_slugs_to_schedules(schedules) if schedules else []
got:
TypeError: create_deployment_schedule_create() got an unexpected keyword argument 'slug'Marvin
07/29/2025, 4:48 PMMarvin
07/29/2025, 4:49 PMTypeError: create_deployment_schedule_create() got an unexpected keyword argument 'slug'
suggests that the schedule creation method doesn't accept a slug parameter in the way you're trying to use it.
In Prefect 3.x, schedules should be defined directly without the need for explicit slugs. Here's how you should structure your schedules instead:
python
# For Cron schedules
schedules = [
{"cron": "0 0 * * *"},
{"cron": "0 12 * * *"}
]
# For Interval schedules
schedules = [
{"interval": "1h"},
{"interval": "30m"}
]
# For RRule schedules
schedules = [
{"rrule": "FREQ=DAILY;INTERVAL=1"},
{"rrule": "FREQ=WEEKLY;BYDAY=MO,WE,FR"}
]
If you're creating a deployment programmatically, you can do:
python
from prefect import flow
from prefect.deployments import Deployment
@flow
def my_flow():
pass
deployment = Deployment.build_from_flow(
flow=my_flow,
name="my-deployment",
schedules=[
{"cron": "0 0 * * *"},
{"interval": "1h"}
]
)
The error you're seeing suggests that you might be trying to add the schedule in a way that's not compatible with Prefect 3.x's API. Could you share:
1. How are you creating your deployment?
2. What type of schedules are you trying to set up (cron, interval, or rrule)?
This will help me provide more specific guidance on how to properly structure your schedules without needing the slug workaround.Dvir Reingewirtz
07/29/2025, 4:55 PMschedules = add_slugs_to_schedules(schedules) if schedules else []
afterwards:
entry_flow.deploy(
name=name,
parameters=default_parameters,
tags=tags,
schedules=schedules,
work_pool_name=work_pool_name,
work_queue_name=work_queue_name,
job_variables=job_variables,
paused=disable_schedule,
image=image,
build=False,
push=False,
)
return None
2. cronMarvin
07/29/2025, 4:55 PMMarvin
07/29/2025, 4:56 PMflow.deploy()
method you're using isn't the recommended approach in 3.x. Instead, you should use flow.from_source().deploy()
. Here's how you should modify your code:
python
# Assuming your schedules are cron expressions
schedules = [
{"cron": cron_expression} for cron_expression in schedules
]
# Create deployment using the new pattern
deployment = entry_flow.from_source(
source=".", # Or your actual source path/URL
entrypoint="path/to/your/flow:entry_flow" # Adjust this to your actual entrypoint
).deploy(
name=name,
parameters=default_parameters,
tags=tags,
schedules=schedules, # Pass schedules directly
work_pool_name=work_pool_name,
work_queue_name=work_queue_name,
job_variables=job_variables,
paused=disable_schedule,
)
The key changes are:
1. Use flow.from_source()
to specify where your flow code lives
2. Chain .deploy()
to create the deployment
3. Pass the schedules directly as a list of dictionaries with the "cron" key
Could you share:
1. What your schedules look like before trying to add slugs?
2. Are you deploying from a local directory or a remote repository?
This will help me provide more specific guidance on how to structure your deployment correctly in Prefect 3.x.Brendan Dalpe
07/30/2025, 5:52 PMDvir Reingewirtz
07/30/2025, 5:58 PMcron
field instead of passing a list in the schedules
field of the flow.deploy()
function.
Thanks for the follow-up!