`prefect.orion.utilities.schemas` ¶

Utilities for creating and working with Orion API schemas.

`IDBaseModel` `pydantic-model` ¶

A PrefectBaseModel with an auto-generated UUID ID value.

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

Source code in prefect/orion/utilities/schemas.py

class IDBaseModel(PrefectBaseModel):
    """
    A PrefectBaseModel with an auto-generated UUID ID value.

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

    id: UUID = Field(default_factory=uuid4)

    def _reset_fields(self) -> Set[str]:
        return super()._reset_fields().union({"id"})

`ORMBaseModel` `pydantic-model` ¶

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 prefect/orion/utilities/schemas.py

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.
    """

    class Config:
        orm_mode = True

    created: DateTimeTZ = Field(None, repr=False)
    updated: DateTimeTZ = Field(None, repr=False)

    def _reset_fields(self) -> Set[str]:
        return super()._reset_fields().union({"created", "updated"})

`PrefectBaseModel` `pydantic-model` ¶

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 prefect/orion/utilities/schemas.py

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.
    """

    class Config:
        # extra attributes are forbidden in order to raise meaningful errors for
        # bad API payloads
        if os.getenv("PREFECT_TEST_MODE"):
            extra = "forbid"
        else:
            extra = "ignore"

        # prevent Pydantic from copying nested models on
        # validation, otherwise ORMBaseModel.copy() is run
        # which resets fields like `id`
        # https://github.com/samuelcolvin/pydantic/pull/2193
        # TODO: remove once this is the default in pydantic>=2.0
        copy_on_model_validation = False

    @classmethod
    def subclass(
        cls: Type[B],
        name: str = None,
        include_fields: List[str] = None,
        exclude_fields: List[str] = None,
    ) -> Type[B]:
        """Creates a subclass of this model containing only the specified fields.

        See `pydantic_subclass()`.

        Args:
            name (str, optional): a name for the subclass
            include_fields (List[str], optional): fields to include
            exclude_fields (List[str], optional): fields to exclude

        Returns:
            BaseModel: a subclass of this class
        """
        return pydantic_subclass(
            base=cls,
            name=name,
            include_fields=include_fields,
            exclude_fields=exclude_fields,
        )

    def _reset_fields(self) -> Set[str]:
        """A set of field names that are reset when the PrefectBaseModel is copied.
        These fields are also disregarded for equality comparisons.
        """
        return set()

    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.dict(exclude=self._reset_fields())
        if isinstance(other, PrefectBaseModel):
            return copy_dict == other.dict(exclude=other._reset_fields())
        if isinstance(other, BaseModel):
            return copy_dict == other.dict()
        else:
            return copy_dict == other

    def json(self, *args, include_secrets: bool = False, **kwargs) -> str:
        """
        Returns a representation of the model as JSON.

        If `include_secrets=True`, then `SecretStr` and `SecretBytes` objects are
        fully revealed. Otherwise they are obfuscated.

        """
        if include_secrets:
            if "encoder" in kwargs:
                raise ValueError(
                    "Alternative encoder provided; can not set encoder for SecretStr and SecretBytes."
                )
            kwargs["encoder"] = partial(
                custom_pydantic_encoder,
                {
                    SecretStr: lambda v: v.get_secret_value() if v else None,
                    SecretBytes: lambda v: v.get_secret_value() if v else None,
                },
            )

        return super().json(*args, **kwargs)

    def dict(
        self, *args, shallow: bool = False, json_compatible: bool = False, **kwargs
    ) -> dict:
        """Returns a representation of the model as a Python dictionary.

        For more information on this distinction please see
        https://pydantic-docs.helpmanual.io/usage/exporting_models/#dictmodel-and-iteration


        Args:
            shallow (bool, optional): If True (default), nested Pydantic fields
                are also coerced to dicts. If false, they are left as Pydantic
                models.
            json_compatible (bool, optional): if True, objects are converted
                into json-compatible representations, similar to calling
                `json.loads(self.json())`. Not compatible with shallow=True.

        Returns:
            dict
        """

        if json_compatible and shallow:
            raise ValueError(
                "`json_compatible` can only be applied to the entire object."
            )

        # return a json-compatible representation of the object
        elif json_compatible:
            return json.loads(self.json(*args, **kwargs))

        # if shallow wasn't requested, return the standard pydantic behavior
        elif not shallow:
            return super().dict(*args, **kwargs)

        # if no options were requested, return simple dict transformation
        # to apply shallow conversion
        elif not args and not kwargs:
            return dict(self)

        # if options like include/exclude were provided, perform
        # a full dict conversion then overwrite with any shallow
        # differences
        else:
            deep_dict = super().dict(*args, **kwargs)
            shallow_dict = dict(self)
            for k, v in list(deep_dict.items()):
                if isinstance(v, dict) and isinstance(shallow_dict[k], BaseModel):
                    deep_dict[k] = shallow_dict[k]
            return deep_dict

    def copy(
        self: T, *, update: Dict = None, reset_fields: bool = False, **kwargs: Any
    ) -> T:
        """
        Duplicate a model.

        Args:
            update: values to change/add to the model copy
            reset_fields: if True, reset the fields specified in `self._reset_fields`
                to their default value on the new model
            kwargs: kwargs to pass to `pydantic.BaseModel.copy`

        Returns:
            A new copy of the model
        """
        if reset_fields:
            update = update or dict()
            for field in self._reset_fields():
                update.setdefault(field, self.__fields__[field].get_default())
        return super().copy(update=update, **kwargs)

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

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

            yield name, value, field.get_default()

`PrefectBaseModel.eq` `special` ¶

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 prefect/orion/utilities/schemas.py

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.dict(exclude=self._reset_fields())
    if isinstance(other, PrefectBaseModel):
        return copy_dict == other.dict(exclude=other._reset_fields())
    if isinstance(other, BaseModel):
        return copy_dict == other.dict()
    else:
        return copy_dict == other

`PrefectBaseModel.copy` ¶

Duplicate a model.

Parameters:

Name	Description	Default
`update`	values to change/add to the model copy `Dict`	`None`
`reset_fields`	if True, reset the fields specified in `self._reset_fields` to their default value on the new model `bool`	`False`
`kwargs`	kwargs to pass to `pydantic.BaseModel.copy` `Any`	`{}`

Returns:

Type	Description
`~T`	A new copy of the model

Source code in prefect/orion/utilities/schemas.py

def copy(
    self: T, *, update: Dict = None, reset_fields: bool = False, **kwargs: Any
) -> T:
    """
    Duplicate a model.

    Args:
        update: values to change/add to the model copy
        reset_fields: if True, reset the fields specified in `self._reset_fields`
            to their default value on the new model
        kwargs: kwargs to pass to `pydantic.BaseModel.copy`

    Returns:
        A new copy of the model
    """
    if reset_fields:
        update = update or dict()
        for field in self._reset_fields():
            update.setdefault(field, self.__fields__[field].get_default())
    return super().copy(update=update, **kwargs)

`PrefectBaseModel.dict` ¶

Returns a representation of the model as a Python dictionary.

For more information on this distinction please see https://pydantic-docs.helpmanual.io/usage/exporting_models/#dictmodel-and-iteration

Parameters:

Name	Description	Default
`shallow`	If True (default), nested Pydantic fields are also coerced to dicts. If false, they are left as Pydantic models. `bool`	`False`
`json_compatible`	if True, objects are converted into json-compatible representations, similar to calling `json.loads(self.json())`. Not compatible with shallow=True. `bool`	`False`

Returns:

Type	Description
`dict`	dict

Source code in prefect/orion/utilities/schemas.py

def dict(
    self, *args, shallow: bool = False, json_compatible: bool = False, **kwargs
) -> dict:
    """Returns a representation of the model as a Python dictionary.

    For more information on this distinction please see
    https://pydantic-docs.helpmanual.io/usage/exporting_models/#dictmodel-and-iteration


    Args:
        shallow (bool, optional): If True (default), nested Pydantic fields
            are also coerced to dicts. If false, they are left as Pydantic
            models.
        json_compatible (bool, optional): if True, objects are converted
            into json-compatible representations, similar to calling
            `json.loads(self.json())`. Not compatible with shallow=True.

    Returns:
        dict
    """

    if json_compatible and shallow:
        raise ValueError(
            "`json_compatible` can only be applied to the entire object."
        )

    # return a json-compatible representation of the object
    elif json_compatible:
        return json.loads(self.json(*args, **kwargs))

    # if shallow wasn't requested, return the standard pydantic behavior
    elif not shallow:
        return super().dict(*args, **kwargs)

    # if no options were requested, return simple dict transformation
    # to apply shallow conversion
    elif not args and not kwargs:
        return dict(self)

    # if options like include/exclude were provided, perform
    # a full dict conversion then overwrite with any shallow
    # differences
    else:
        deep_dict = super().dict(*args, **kwargs)
        shallow_dict = dict(self)
        for k, v in list(deep_dict.items()):
            if isinstance(v, dict) and isinstance(shallow_dict[k], BaseModel):
                deep_dict[k] = shallow_dict[k]
        return deep_dict

`PrefectBaseModel.json` ¶

Returns a representation of the model as JSON.

If include_secrets=True, then SecretStr and SecretBytes objects are fully revealed. Otherwise they are obfuscated.

Source code in prefect/orion/utilities/schemas.py

def json(self, *args, include_secrets: bool = False, **kwargs) -> str:
    """
    Returns a representation of the model as JSON.

    If `include_secrets=True`, then `SecretStr` and `SecretBytes` objects are
    fully revealed. Otherwise they are obfuscated.

    """
    if include_secrets:
        if "encoder" in kwargs:
            raise ValueError(
                "Alternative encoder provided; can not set encoder for SecretStr and SecretBytes."
            )
        kwargs["encoder"] = partial(
            custom_pydantic_encoder,
            {
                SecretStr: lambda v: v.get_secret_value() if v else None,
                SecretBytes: lambda v: v.get_secret_value() if v else None,
            },
        )

    return super().json(*args, **kwargs)

`PrefectBaseModel.subclass` `classmethod` ¶

Creates a subclass of this model containing only the specified fields.

See pydantic_subclass().

Parameters:

Name	Description	Default
`name`	a name for the subclass `str`	`None`
`include_fields`	fields to include `List[str]`	`None`
`exclude_fields`	fields to exclude `List[str]`	`None`

Returns:

Type	Description
`BaseModel`	a subclass of this class

Source code in prefect/orion/utilities/schemas.py

@classmethod
def subclass(
    cls: Type[B],
    name: str = None,
    include_fields: List[str] = None,
    exclude_fields: List[str] = None,
) -> Type[B]:
    """Creates a subclass of this model containing only the specified fields.

    See `pydantic_subclass()`.

    Args:
        name (str, optional): a name for the subclass
        include_fields (List[str], optional): fields to include
        exclude_fields (List[str], optional): fields to exclude

    Returns:
        BaseModel: a subclass of this class
    """
    return pydantic_subclass(
        base=cls,
        name=name,
        include_fields=include_fields,
        exclude_fields=exclude_fields,
    )

`pydantic_subclass` ¶

Creates a subclass of a Pydantic model that excludes certain fields. Pydantic models use the fields attribute of their parent class to determine inherited fields, so to create a subclass without fields, we temporarily remove those fields from the parent fields and use create_model to dynamically generate a new subclass.

Parameters:

Name	Description	Default
`base`	a Pydantic BaseModel `pydantic.BaseModel`	required
`name`	a name for the subclass. If not provided it will have the same name as the base class. `str`	`None`
`include_fields`	a set of field names to include. If `None`, all fields are included. `List[str]`	`None`
`exclude_fields`	a list of field names to exclude. If `None`, no fields are excluded. `List[str]`	`None`

Returns:

Type	Description
`pydantic.BaseModel`	a new model subclass that contains only the specified fields.

Examples:

To subclass a model with a subset of fields:

class Parent(pydantic.BaseModel):
    x: int = 1
    y: int = 2

Child = pydantic_subclass(Parent, 'Child', exclude_fields=['y'])
assert hasattr(Child(), 'x')
assert not hasattr(Child(), 'y')

To subclass a model with a subset of fields but include a new field:

class Child(pydantic_subclass(Parent, exclude_fields=['y'])):
    z: int = 3

assert hasattr(Child(), 'x')
assert not hasattr(Child(), 'y')
assert hasattr(Child(), 'z')

Source code in prefect/orion/utilities/schemas.py

def pydantic_subclass(
    base: Type[B],
    name: str = None,
    include_fields: List[str] = None,
    exclude_fields: List[str] = None,
) -> Type[B]:
    """Creates a subclass of a Pydantic model that excludes certain fields.
    Pydantic models use the __fields__ attribute of their parent class to
    determine inherited fields, so to create a subclass without fields, we
    temporarily remove those fields from the parent __fields__ and use
    `create_model` to dynamically generate a new subclass.

    Args:
        base (pydantic.BaseModel): a Pydantic BaseModel
        name (str): a name for the subclass. If not provided
            it will have the same name as the base class.
        include_fields (List[str]): a set of field names to include.
            If `None`, all fields are included.
        exclude_fields (List[str]): a list of field names to exclude.
            If `None`, no fields are excluded.

    Returns:
        pydantic.BaseModel: a new model subclass that contains only the specified fields.

    Example:
        To subclass a model with a subset of fields:
        ```python
        class Parent(pydantic.BaseModel):
            x: int = 1
            y: int = 2

        Child = pydantic_subclass(Parent, 'Child', exclude_fields=['y'])
        assert hasattr(Child(), 'x')
        assert not hasattr(Child(), 'y')
        ```

        To subclass a model with a subset of fields but include a new field:
        ```python
        class Child(pydantic_subclass(Parent, exclude_fields=['y'])):
            z: int = 3

        assert hasattr(Child(), 'x')
        assert not hasattr(Child(), 'y')
        assert hasattr(Child(), 'z')
        ```
    """

    # collect field names
    field_names = set(include_fields or base.__fields__)
    excluded_fields = set(exclude_fields or [])
    if field_names.difference(base.__fields__):
        raise ValueError(
            "Included fields not found on base class: "
            f"{field_names.difference(base.__fields__)}"
        )
    elif excluded_fields.difference(base.__fields__):
        raise ValueError(
            "Excluded fields not found on base class: "
            f"{excluded_fields.difference(base.__fields__)}"
        )
    field_names.difference_update(excluded_fields)

    # create a new class that inherits from `base` but only contains the specified
    # pydantic __fields__
    new_cls = type(
        name or base.__name__,
        (base,),
        {
            "__fields__": {
                k: copy.copy(v) for k, v in base.__fields__.items() if k in field_names
            }
        },
    )

    return new_cls

prefect.orion.utilities.schemas ¶

`prefect.orion.utilities.schemas` ¶