File: //lib/python3/dist-packages/cbor2/_types.py
from __future__ import annotations
import threading
from collections import namedtuple
from collections.abc import Iterable, Iterator
from functools import total_ordering
from reprlib import recursive_repr
from typing import Any, Mapping, TypeVar
KT = TypeVar("KT")
VT_co = TypeVar("VT_co", covariant=True)
thread_locals = threading.local()
class CBORError(Exception):
"""Base class for errors that occur during CBOR encoding or decoding."""
class CBOREncodeError(CBORError):
"""Raised for exceptions occurring during CBOR encoding."""
class CBOREncodeTypeError(CBOREncodeError, TypeError):
"""Raised when attempting to encode a type that cannot be serialized."""
class CBOREncodeValueError(CBOREncodeError, ValueError):
"""Raised when the CBOR encoder encounters an invalid value."""
class CBORDecodeError(CBORError):
"""Raised for exceptions occurring during CBOR decoding."""
class CBORDecodeValueError(CBORDecodeError, ValueError):
"""Raised when the CBOR stream being decoded contains an invalid value."""
class CBORDecodeEOF(CBORDecodeError, EOFError):
"""Raised when decoding unexpectedly reaches EOF."""
@total_ordering
class CBORTag:
"""
Represents a CBOR semantic tag.
:param int tag: tag number
:param value: encapsulated value (any object)
"""
__slots__ = "tag", "value"
def __init__(self, tag: str | int, value: Any) -> None:
if not isinstance(tag, int) or tag not in range(2**64):
raise TypeError("CBORTag tags must be positive integers less than 2**64")
self.tag = tag
self.value = value
def __eq__(self, other: object) -> bool:
if isinstance(other, CBORTag):
return (self.tag, self.value) == (other.tag, other.value)
return NotImplemented
def __le__(self, other: object) -> bool:
if isinstance(other, CBORTag):
return (self.tag, self.value) <= (other.tag, other.value)
return NotImplemented
@recursive_repr()
def __repr__(self) -> str:
return f"CBORTag({self.tag}, {self.value!r})"
def __hash__(self) -> int:
self_id = id(self)
try:
running_hashes = thread_locals.running_hashes
except AttributeError:
running_hashes = thread_locals.running_hashes = set()
if self_id in running_hashes:
raise RuntimeError(
"This CBORTag is not hashable because it contains a reference to itself"
)
running_hashes.add(self_id)
try:
return hash((self.tag, self.value))
finally:
running_hashes.remove(self_id)
if not running_hashes:
del thread_locals.running_hashes
class CBORSimpleValue(namedtuple("CBORSimpleValue", ["value"])):
"""
Represents a CBOR "simple value".
:param int value: the value (0-255)
"""
__slots__ = ()
value: int
def __hash__(self) -> int:
return hash(self.value)
def __new__(cls, value: int) -> CBORSimpleValue:
if value < 0 or value > 255 or 23 < value < 32:
raise TypeError("simple value out of range (0..23, 32..255)")
return super().__new__(cls, value)
def __eq__(self, other: object) -> bool:
if isinstance(other, int):
return self.value == other
elif isinstance(other, CBORSimpleValue):
return self.value == other.value
return NotImplemented
def __ne__(self, other: object) -> bool:
if isinstance(other, int):
return self.value != other
elif isinstance(other, CBORSimpleValue):
return self.value != other.value
return NotImplemented
def __lt__(self, other: object) -> bool:
if isinstance(other, int):
return self.value < other
elif isinstance(other, CBORSimpleValue):
return self.value < other.value
return NotImplemented
def __le__(self, other: object) -> bool:
if isinstance(other, int):
return self.value <= other
elif isinstance(other, CBORSimpleValue):
return self.value <= other.value
return NotImplemented
def __ge__(self, other: object) -> bool:
if isinstance(other, int):
return self.value >= other
elif isinstance(other, CBORSimpleValue):
return self.value >= other.value
return NotImplemented
def __gt__(self, other: object) -> bool:
if isinstance(other, int):
return self.value > other
elif isinstance(other, CBORSimpleValue):
return self.value > other.value
return NotImplemented
class FrozenDict(Mapping[KT, VT_co]):
"""
A hashable, immutable mapping type.
The arguments to ``FrozenDict`` are processed just like those to ``dict``.
"""
def __init__(self, *args: Mapping[KT, VT_co] | Iterable[tuple[KT, VT_co]]) -> None:
self._d: dict[KT, VT_co] = dict(*args)
self._hash: int | None = None
def __iter__(self) -> Iterator[KT]:
return iter(self._d)
def __len__(self) -> int:
return len(self._d)
def __getitem__(self, key: KT) -> VT_co:
return self._d[key]
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self._d})"
def __hash__(self) -> int:
if self._hash is None:
self._hash = hash((frozenset(self), frozenset(self.values())))
return self._hash
class UndefinedType:
__slots__ = ()
def __new__(cls: type[UndefinedType]) -> UndefinedType:
try:
return undefined
except NameError:
return super().__new__(cls)
def __repr__(self) -> str:
return "undefined"
def __bool__(self) -> bool:
return False
class BreakMarkerType:
__slots__ = ()
def __new__(cls: type[BreakMarkerType]) -> BreakMarkerType:
try:
return break_marker
except NameError:
return super().__new__(cls)
def __repr__(self) -> str:
return "break_marker"
def __bool__(self) -> bool:
return True
#: Represents the "undefined" value.
undefined = UndefinedType()
break_marker = BreakMarkerType()