<@ULVA73B9P> im getting this error after migrating...
# marvin-ai
d
@Marvin im getting this error after migrating from prefect 2 to 3: Response: {'detail': 'Please provide a slug for each schedule in your request to ensure schedules are updated correctly.'} is slug a mandatory?
m
thought for 39.8 seconds
Based on the error message you're receiving and my research, it seems you're encountering a migration-related issue. The error suggests that the API is expecting a slug for each schedule, which wasn't necessarily required in Prefect 2.x. Here's what you need to know: 1. In Prefect 3.x, when updating or creating schedules, you should provide a unique identifier (slug) for each schedule. This helps ensure that schedules are updated correctly, especially when dealing with multiple schedules for a single deployment. 2. The slug should be a unique string identifier for your schedule. For example:
Copy code
python
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:
Copy code
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.
d
@Marvin tried add it behind the scene as follows:
Copy code
def 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'
m
thought for 70.8 seconds
Looking at your code and the error message, I can see what's happening. The approach you're taking isn't quite right for Prefect 3.x. The error
TypeError: 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:
Copy code
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:
Copy code
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.
d
@Marvin 1. using deployment as follows: adding schedules as shown above
Copy code
schedules = add_slugs_to_schedules(schedules) if schedules else []
afterwards:
Copy code
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. cron
m
thought for 75.5 seconds
I apologize for my earlier response. I see now that there's a fundamental difference in how deployments work in Prefect 3.x. The direct
flow.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:
Copy 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.
b
Hey @Dvir Reingewirtz did you get this sorted?
d
Hi @Brendan Dalpe, Yes, I eventually solved the issue by passing a string in the
cron
field instead of passing a list in the
schedules
field of the
flow.deploy()
function. Thanks for the follow-up!
🙌 1