Skip to content

prefect.server.utilities.schemas.bases

IDBaseModel

Bases: PrefectBaseModel

A PrefectBaseModel with an auto-generated UUID ID value.

The ID is reset on copy() and not included in equality comparisons.

Source code in src/prefect/server/utilities/schemas/bases.py 
172
173
174
175
176
177
178
179
180
class IDBaseModel(PrefectBaseModel):
    """
    A PrefectBaseModel with an auto-generated UUID ID value.

    The ID is reset on copy() and not included in equality comparisons.
    """

    _reset_fields: ClassVar[Set[str]] = {"id"}
    id: UUID = Field(default_factory=uuid4)

ORMBaseModel

Bases: IDBaseModel

A PrefectBaseModel with an auto-generated UUID ID value and created / updated timestamps, intended for compatibility with our standard ORM models.

The ID, created, and updated fields are reset on copy() and not included in equality comparisons.

Source code in src/prefect/server/utilities/schemas/bases.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class ORMBaseModel(IDBaseModel):
    """
    A PrefectBaseModel with an auto-generated UUID ID value and created /
    updated timestamps, intended for compatibility with our standard ORM models.

    The ID, created, and updated fields are reset on copy() and not included in
    equality comparisons.
    """

    _reset_fields: ClassVar[Set[str]] = {"id", "created", "updated"}

    model_config = ConfigDict(from_attributes=True)

    created: Optional[DateTime] = Field(default=None, repr=False)
    updated: Optional[DateTime] = Field(default=None, repr=False)

PrefectBaseModel

Bases: BaseModel

A base pydantic.BaseModel for all Prefect schemas and pydantic models.

As the basis for most Prefect schemas, this base model usually ignores extra fields that are passed to it at instantiation. Because adding new fields to API payloads is not considered a breaking change, this ensures that any Prefect client loading data from a server running a possibly-newer version of Prefect will be able to process those new fields gracefully. However, when PREFECT_TEST_MODE is on, extra fields are forbidden in order to catch subtle unintentional testing errors.

Source code in src/prefect/server/utilities/schemas/bases.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class PrefectBaseModel(BaseModel):
    """A base pydantic.BaseModel for all Prefect schemas and pydantic models.

    As the basis for most Prefect schemas, this base model usually ignores extra
    fields that are passed to it at instantiation. Because adding new fields to
    API payloads is not considered a breaking change, this ensures that any
    Prefect client loading data from a server running a possibly-newer version
    of Prefect will be able to process those new fields gracefully. However,
    when PREFECT_TEST_MODE is on, extra fields are forbidden in order to catch
    subtle unintentional testing errors.
    """

    _reset_fields: ClassVar[Set[str]] = set()
    model_config = ConfigDict(
        ser_json_timedelta="float",
        extra=(
            "ignore"
            if os.getenv("PREFECT_TEST_MODE", "0").lower() not in ["true", "1"]
            else "forbid"
        ),
    )

    def __eq__(self, other: Any) -> bool:
        """Equaltiy operator that ignores the resettable fields of the PrefectBaseModel.

        NOTE: this equality operator will only be applied if the PrefectBaseModel is
        the left-hand operand. This is a limitation of Python.
        """
        copy_dict = self.model_dump(exclude=self._reset_fields)
        if isinstance(other, PrefectBaseModel):
            return copy_dict == other.model_dump(exclude=other._reset_fields)
        if isinstance(other, BaseModel):
            return copy_dict == other.model_dump()
        else:
            return copy_dict == other

    def __rich_repr__(self):
        # Display all of the fields in the model if they differ from the default value
        for name, field in self.model_fields.items():
            value = getattr(self, name)

            # Simplify the display of some common fields
            if field.annotation == UUID and value:
                value = str(value)
            elif (
                isinstance(field.annotation, datetime.datetime)
                and name == "timestamp"
                and value
            ):
                value = pendulum.instance(value).isoformat()
            elif isinstance(field.annotation, datetime.datetime) and value:
                value = pendulum.instance(value).diff_for_humans()

            yield name, value, field.get_default()

    def reset_fields(self: Self) -> Self:
        """
        Reset the fields of the model that are in the `_reset_fields` set.

        Returns:
            PrefectBaseModel: A new instance of the model with the reset fields.
        """
        return self.model_copy(
            update={
                field: self.model_fields[field].get_default(call_default_factory=True)
                for field in self._reset_fields
            }
        )

    def model_dump_for_orm(
        self,
        *,
        include: "IncEx" = None,
        exclude: "IncEx" = None,
        by_alias: bool = False,
        exclude_unset: bool = False,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
    ) -> Dict[str, Any]:
        """
        Prefect extension to `BaseModel.model_dump`.  Generate a Python dictionary
        representation of the model suitable for passing to SQLAlchemy model
        constructors, `INSERT` statements, etc.  The critical difference here is that
        this method will return any nested BaseModel objects as `BaseModel` instances,
        rather than serialized Python dictionaries.

        Accepts the standard Pydantic `model_dump` arguments, except for `mode` (which
        is always "python"), `round_trip`, and `warnings`.

        Usage docs: https://docs.pydantic.dev/2.6/concepts/serialization/#modelmodel_dump

        Args:
            include: A list of fields to include in the output.
            exclude: A list of fields to exclude from the output.
            by_alias: Whether to use the field's alias in the dictionary key if defined.
            exclude_unset: Whether to exclude fields that have not been explicitly set.
            exclude_defaults: Whether to exclude fields that are set to their default
                value.
            exclude_none: Whether to exclude fields that have a value of `None`.

        Returns:
            A dictionary representation of the model, suitable for passing
            to SQLAlchemy model constructors, INSERT statements, etc.
        """
        # TODO: this could be optimized by excluding any fields that we know we are
        # going to replace because they are `BaseModel` instances.  This would involve
        # understanding which fields would be included or excluded by model_dump so we
        # could instruct Pydantic to exclude them up front.
        deep = self.model_dump(
            mode="python",
            include=include,
            exclude=exclude,
            by_alias=by_alias,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
            context={"for_orm": True},
        )
        for k, v in self:
            if k in deep and isinstance(v, BaseModel):
                deep[k] = v
        return deep

__eq__(other)

Equaltiy operator that ignores the resettable fields of the PrefectBaseModel.

NOTE: this equality operator will only be applied if the PrefectBaseModel is the left-hand operand. This is a limitation of Python.

Source code in src/prefect/server/utilities/schemas/bases.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
def __eq__(self, other: Any) -> bool:
    """Equaltiy operator that ignores the resettable fields of the PrefectBaseModel.

    NOTE: this equality operator will only be applied if the PrefectBaseModel is
    the left-hand operand. This is a limitation of Python.
    """
    copy_dict = self.model_dump(exclude=self._reset_fields)
    if isinstance(other, PrefectBaseModel):
        return copy_dict == other.model_dump(exclude=other._reset_fields)
    if isinstance(other, BaseModel):
        return copy_dict == other.model_dump()
    else:
        return copy_dict == other

model_dump_for_orm(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False)

Prefect extension to BaseModel.model_dump. Generate a Python dictionary representation of the model suitable for passing to SQLAlchemy model constructors, INSERT statements, etc. The critical difference here is that this method will return any nested BaseModel objects as BaseModel instances, rather than serialized Python dictionaries.

Accepts the standard Pydantic model_dump arguments, except for mode (which is always "python"), round_trip, and warnings.

Usage docs: https://docs.pydantic.dev/2.6/concepts/serialization/#modelmodel_dump

Parameters:

Name Type Description Default
include IncEx

A list of fields to include in the output.

 None
exclude IncEx

A list of fields to exclude from the output.

 None
by_alias bool

Whether to use the field's alias in the dictionary key if defined.

 False
exclude_unset bool

Whether to exclude fields that have not been explicitly set.

 False
exclude_defaults bool

Whether to exclude fields that are set to their default value.

 False
exclude_none bool

Whether to exclude fields that have a value of None.

 False

Returns:

Type Description
Dict[str, Any]

A dictionary representation of the model, suitable for passing
Dict[str, Any]

to SQLAlchemy model constructors, INSERT statements, etc.
Source code in src/prefect/server/utilities/schemas/bases.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def model_dump_for_orm(
    self,
    *,
    include: "IncEx" = None,
    exclude: "IncEx" = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
) -> Dict[str, Any]:
    """
    Prefect extension to `BaseModel.model_dump`.  Generate a Python dictionary
    representation of the model suitable for passing to SQLAlchemy model
    constructors, `INSERT` statements, etc.  The critical difference here is that
    this method will return any nested BaseModel objects as `BaseModel` instances,
    rather than serialized Python dictionaries.

    Accepts the standard Pydantic `model_dump` arguments, except for `mode` (which
    is always "python"), `round_trip`, and `warnings`.

    Usage docs: https://docs.pydantic.dev/2.6/concepts/serialization/#modelmodel_dump

    Args:
        include: A list of fields to include in the output.
        exclude: A list of fields to exclude from the output.
        by_alias: Whether to use the field's alias in the dictionary key if defined.
        exclude_unset: Whether to exclude fields that have not been explicitly set.
        exclude_defaults: Whether to exclude fields that are set to their default
            value.
        exclude_none: Whether to exclude fields that have a value of `None`.

    Returns:
        A dictionary representation of the model, suitable for passing
        to SQLAlchemy model constructors, INSERT statements, etc.
    """
    # TODO: this could be optimized by excluding any fields that we know we are
    # going to replace because they are `BaseModel` instances.  This would involve
    # understanding which fields would be included or excluded by model_dump so we
    # could instruct Pydantic to exclude them up front.
    deep = self.model_dump(
        mode="python",
        include=include,
        exclude=exclude,
        by_alias=by_alias,
        exclude_unset=exclude_unset,
        exclude_defaults=exclude_defaults,
        exclude_none=exclude_none,
        context={"for_orm": True},
    )
    for k, v in self:
        if k in deep and isinstance(v, BaseModel):
            deep[k] = v
    return deep

reset_fields()

Reset the fields of the model that are in the _reset_fields set.

Returns:

Name Type Description
PrefectBaseModel Self

A new instance of the model with the reset fields.
Source code in src/prefect/server/utilities/schemas/bases.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
def reset_fields(self: Self) -> Self:
    """
    Reset the fields of the model that are in the `_reset_fields` set.

    Returns:
        PrefectBaseModel: A new instance of the model with the reset fields.
    """
    return self.model_copy(
        update={
            field: self.model_fields[field].get_default(call_default_factory=True)
            for field in self._reset_fields
        }
    )

get_class_fields_only(model)

Gets all the field names defined on the model class but not any parent classes. Any fields that are on the parent but redefined on the subclass are included.

Source code in src/prefect/server/utilities/schemas/bases.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def get_class_fields_only(model: Type[BaseModel]) -> set:
    """
    Gets all the field names defined on the model class but not any parent classes.
    Any fields that are on the parent but redefined on the subclass are included.
    """
    subclass_class_fields = set(model.__annotations__.keys())
    parent_class_fields = set()

    for base in model.__class__.__bases__:
        if issubclass(base, BaseModel):
            parent_class_fields.update(base.__annotations__.keys())

    return (subclass_class_fields - parent_class_fields) | (
        subclass_class_fields & parent_class_fields
    )