from typing import Annotated, Any, Callable, TypeVar
from pydantic import (
Field,
FieldSerializationInfo,
ValidationInfo,
WrapSerializer,
WrapValidator,
)
__all__ = ("PiperUndefined", "PIPER_UNDEFINED")
_T = TypeVar("Generic Type")
[docs]
class PiperUndefined:
"""
Sentinel class to represent an uninitialized configuration parameter.
In the Piper framework, developers must provide default values so that
configurations can be dumped to YAML. However, standard defaults like
``None`` are often valid configuration values themselves.
This class provides a unique sentinel, `PIPER_UNDEFINED`, to explicitly
mark parameters that must be provided by the user at runtime,
distinguishing them from both ``None`` and valid data.
Attributes
----------
__json_repr__ : str
The string representation used when serializing the sentinel to JSON.
"""
__json_repr__ = "!__PIPER_UNDEFINED__!"
[docs]
@staticmethod
def check(value: Any) -> bool:
"""
Check if a value is undefined,
either by matching the JSON representation or
being the ``PIPER_UNDEFINED`` singleton.
Parameters
----------
value
The value to check.
Returns
-------
``True`` if value is equivalent of ``PIPER_UNDEFINED``,
otherwise ``False`` .
"""
return (value == PiperUndefined.__json_repr__) or (
value is PIPER_UNDEFINED
)
[docs]
@staticmethod
def validate(value: Any, handler: Callable, info: ValidationInfo) -> Any:
"""
Validate if the value is the ``PIPER_UNDEFINED`` sentinel.
If the validation context doesn't allow ``PIPER_UNDEFINED`` values,
then raise exceptions.
This must be used as a "wrap"-type validator for a pydantic field.
Parameters
----------
value
The value to be validated.
handler
The next validation handler in the Pydantic chain.
info
Pydantic validation context and metadata.
Returns
-------
If value is a equivalent of ``PIPER_UNDEFINED``, then the
``PIPER_UNDEFINED`` sentinel.
Else, the result of the pyndatic's default validation handler.
Raises
------
AssertionError
If the value is ``PIPER_UNDEFINED`` but the validation context
specifically disallows unset fields (``allow_unset=False``).
"""
allow_unset = True
if isinstance(info.context, dict):
allow_unset = info.context.get("allow_unset", True)
if PiperUndefined.check(value):
assert allow_unset, f"Field '{info.field_name}' is not set."
return PIPER_UNDEFINED
return handler(value)
[docs]
@staticmethod
def serialize(
value: Any, handler: Callable, info: FieldSerializationInfo
) -> Any:
"""
Serialize the ``PIPER_UNDEFINED`` sentinel based on the output mode.
If value is
Parameters
----------
value
The value to serialize.
handler
The next serialization handler in the chain.
info
Metadata about the serialization process, including the mode.
Returns
-------
If value is a equivalent of ``PIPER_UNDEFINED``, then either:
- the string "!__PIPER_UNDEFINED__!" if mode is 'json'
- otherwise the ``PIPER_UNDEFINED`` object.
Else, the result of the pyndatic's default serializer handler.
"""
if PiperUndefined.check(value):
if info.mode == "json":
return PiperUndefined.__json_repr__
return PIPER_UNDEFINED
return handler(value)
[docs]
@staticmethod
def annotate(annotation: type[_T]) -> type[_T]:
"""
Wraps a given type annotation with serializer
and deserializer for the ``PIPER_UNDEFINED``.
Also sets the default value to ``PIPER_UNDEFINED``.
If required, caller should ensure that default value is not
overriden by other ways of definining a pydantic field default.
"""
return Annotated[
annotation,
Field(default=PIPER_UNDEFINED),
WrapSerializer(PiperUndefined.serialize),
WrapValidator(PiperUndefined.validate),
]
def __repr__(self) -> str:
"""
Return the string representation of the sentinel.
Returns "<PIPER_UNDEFINED>".
"""
return "<PIPER_UNDEFINED>"
PIPER_UNDEFINED = PiperUndefined()
"""
The singleton instance of the `PiperUndefined` sentinel used across the
Piper framework to denote config parameters with undefined values
while defining the stage.
User must provide values for such parameters while running the pipeline.
"""