More effectively coping with Python's 'type system'

This week, when reviewing a PR, I observed that Guidry had added a new field to the RunDeployment action: schedule_after. This new field allows users to schedule a deployment run some fixed time after a condition is met.

Due to reasons orthogonal to this post, we have at least 3 places we need to update schemas when we add a new field like this:

• the object model itself, that is, the RunDeployment action model
• the client-side schema for RunDeployment
• the server-side schema for RunDeployment

It’s not important why this is true, but the fact of the matter is that we have 3 separate BaseModel subclasses that all will get this new field.

A “delay” must be a positive amount of time, so we want to assert that schedule_after must be a non-negative timedelta. Therefore, as we are pydantic users, the PR included 3 separate but identical @field_validator implementations that checked (for each schema) that the value was non-negative.

This could be fine, but I am also haunted by the experience of migrating from pydantic v1 to v2, which included a great deal of moving these functionally identical validators around on multiple schemas.

Therefore, we’ve begun to use Annotated types more often to bind validation logic to field types themselves, so that we don’t need to repeat said validation logic on multiple schemas that use those types.

So, that’s what I suggested and that’s what we did!

It works out really clean (if I do say so myself), and looks like this:

 1# src/prefect/types/__init__.py
 2
 3from datetime import timedelta
 4from typing import Annotated
 5
 6from pydantic import AfterValidator
 7
 8def _validate_non_negative_timedelta(v: timedelta) -> timedelta:
 9    if v < timedelta(seconds=0):
10        raise ValueError("timedelta must be non-negative")
11    return v
12
13NonNegativeTimedelta = Annotated[
14    timedelta,
15    AfterValidator(_validate_non_negative_timedelta)
16]
17
18# src/prefect/server/schemas/actions.py
19
20from prefect.types import NonNegativeTimedelta
21
22class RunDeployment(BaseModel):
23    # ...
24    schedule_after: NonNegativeTimedelta

… and so on for the rest of the schemas.

The nice things are that:

• you write the validation logic once
• the field types become an interface you can switch out the implementation of
• the field types become self-documenting

Here’s a (entirely unedited 🙃) youtube video that I made about this (checks calendar) almost a couple years ago now: