exceptions.py
python
"""Исключения asyncio."""
__all__ = ('BrokenBarrierError',
'CancelledError', 'InvalidStateError', 'TimeoutError',
'IncompleteReadError', 'LimitOverrunError',
'SendfileNotAvailableError')
class CancelledError(BaseException):
"""Future или задача были отменены."""
TimeoutError = TimeoutError # создать локальный псевдоним для стандартного исключения
class InvalidStateError(Exception):
"""Операция не разрешена в данном состоянии."""
class SendfileNotAvailableError(RuntimeError):
"""Системный вызов sendfile недоступен.
Возникает, если ОС не поддерживает системный вызов sendfile для данного сокета или
типа файла.
"""
class IncompleteReadError(EOFError):
"""
Ошибка неполного чтения. Атрибуты:
- partial: строка байтов, прочитанная до достижения конца потока
- expected: общее количество ожидаемых байтов (или None, если неизвестно)
"""
def __init__(self, partial, expected):
r_expected = 'undefined' if expected is None else repr(expected)
super().__init__(f'{len(partial)} bytes read on a total of '
f'{r_expected} expected bytes')
self.partial = partial
self.expected = expected
def __reduce__(self):
return type(self), (self.partial, self.expected)
class LimitOverrunError(Exception):
"""Достигнут лимит буфера во время поиска разделителя.
Атрибуты:
- consumed: общее количество байтов, которые необходимо потребить.
"""
def __init__(self, message, consumed):
super().__init__(message)
self.consumed = consumed
def __reduce__(self):
return type(self), (self.args[0], self.consumed)
class BrokenBarrierError(RuntimeError):
"""Барьер сломан вызовом barrier.abort()."""