Source code for faust.exceptions

"""Faust exceptions."""

import typing

__all__ = [
    "FaustError",
    "FaustWarning",
    "FaustPredicate",
    "SecurityError",
    "NotReady",
    "Skip",
    "AlreadyConfiguredWarning",
    "ImproperlyConfigured",
    "ValidationError",
    "DecodeError",
    "KeyDecodeError",
    "ValueDecodeError",
    "SameNode",
    "ProducerSendError",
    "ConsumerNotStarted",
    "PartitionsMismatch",
    "ConsistencyError",
]

if typing.TYPE_CHECKING:
    from .types.models import FieldDescriptorT as _FieldDescriptorT
else:

    class _FieldDescriptorT: ...  # noqa



[docs]class FaustError(Exception):
    """Base-class for all Faust exceptions."""




[docs]class FaustWarning(UserWarning):
    """Base-class for all Faust warnings."""




[docs]class FaustPredicate(FaustError):
    """Base-class for semi-predicates such as :exc:`Skip`."""




[docs]class SecurityError(FaustError):
    """Base-class for security related (high priority) exceptions."""




[docs]class NotReady(FaustPredicate):
    """Service not started."""







[docs]class AlreadyConfiguredWarning(FaustWarning):
    """Trying to configure app after configuration accessed."""




[docs]class ImproperlyConfigured(FaustError):
    """The library is not configured/installed correctly."""




[docs]class ValidationError(FaustError, ValueError):
    """Value passed for model field is not valid."""

    field: _FieldDescriptorT

    def __init__(self, reason: str, *, field: _FieldDescriptorT) -> None:
        self.reason = reason
        self.field = field
        super().__init__(reason, field)

    def __str__(self) -> str:
        return f"{self.reason} {self.field!r}"

    def __repr__(self) -> str:
        return f"<{type(self).__name__}: {self}>"




[docs]class DecodeError(FaustError):
    """Error while decoding/deserializing message key/value."""




[docs]class KeyDecodeError(DecodeError):
    """Error while decoding/deserializing message key."""




[docs]class ValueDecodeError(DecodeError):
    """Error while decoding/deserializing message value."""




[docs]class SameNode(FaustPredicate):
    """Exception raised by router when data is located on same node."""




[docs]class ProducerSendError(FaustError):
    """Error while sending attached messages prior to commit."""




[docs]class ConsumerNotStarted(NotReady):
    """Error trying to perform operation on consumer not started."""




[docs]class PartitionsMismatch(FaustError):
    """Number of partitions between related topics differ."""




[docs]class ConsistencyError(FaustError):
    """Persisted table offset is out of sync with changelog topic highwater."""