venv/lib/python3.11/site-packages/pip/_vendor/msgpack/ext.py

   1 # coding: utf-8
   2 from collections import namedtuple
   3 import datetime
   4 import sys
   5 import struct
   6
   7
   8 PY2 = sys.version_info[0] == 2
   9
  10 if PY2:
  11     int_types = (int, long)
  12     _utc = None
  13 else:
  14     int_types = int
  15     try:
  16         _utc = datetime.timezone.utc
  17     except AttributeError:
  18         _utc = datetime.timezone(datetime.timedelta(0))
  19
  20
  21 class ExtType(namedtuple("ExtType", "code data")):
  22     """ExtType represents ext type in msgpack."""
  23
  24     def __new__(cls, code, data):
  25         if not isinstance(code, int):
  26             raise TypeError("code must be int")
  27         if not isinstance(data, bytes):
  28             raise TypeError("data must be bytes")
  29         if not 0 <= code <= 127:
  30             raise ValueError("code must be 0~127")
  31         return super(ExtType, cls).__new__(cls, code, data)
  32
  33
  34 class Timestamp(object):
  35     """Timestamp represents the Timestamp extension type in msgpack.
  36
  37     When built with Cython, msgpack uses C methods to pack and unpack `Timestamp`. When using pure-Python
  38     msgpack, :func:`to_bytes` and :func:`from_bytes` are used to pack and unpack `Timestamp`.
  39
  40     This class is immutable: Do not override seconds and nanoseconds.
  41     """
  42
  43     __slots__ = ["seconds", "nanoseconds"]
  44
  45     def __init__(self, seconds, nanoseconds=0):
  46         """Initialize a Timestamp object.
  47
  48         :param int seconds:
  49             Number of seconds since the UNIX epoch (00:00:00 UTC Jan 1 1970, minus leap seconds).
  50             May be negative.
  51
  52         :param int nanoseconds:
  53             Number of nanoseconds to add to `seconds` to get fractional time.
  54             Maximum is 999_999_999.  Default is 0.
  55
  56         Note: Negative times (before the UNIX epoch) are represented as negative seconds + positive ns.
  57         """
  58         if not isinstance(seconds, int_types):
  59             raise TypeError("seconds must be an integer")
  60         if not isinstance(nanoseconds, int_types):
  61             raise TypeError("nanoseconds must be an integer")
  62         if not (0 <= nanoseconds < 10**9):
  63             raise ValueError(
  64                 "nanoseconds must be a non-negative integer less than 999999999."
  65             )
  66         self.seconds = seconds
  67         self.nanoseconds = nanoseconds
  68
  69     def __repr__(self):
  70         """String representation of Timestamp."""
  71         return "Timestamp(seconds={0}, nanoseconds={1})".format(
  72             self.seconds, self.nanoseconds
  73         )
  74
  75     def __eq__(self, other):
  76         """Check for equality with another Timestamp object"""
  77         if type(other) is self.__class__:
  78             return (
  79                 self.seconds == other.seconds and self.nanoseconds == other.nanoseconds
  80             )
  81         return False
  82
  83     def __ne__(self, other):
  84         """not-equals method (see :func:`__eq__()`)"""
  85         return not self.__eq__(other)
  86
  87     def __hash__(self):
  88         return hash((self.seconds, self.nanoseconds))
  89
  90     @staticmethod
  91     def from_bytes(b):
  92         """Unpack bytes into a `Timestamp` object.
  93
  94         Used for pure-Python msgpack unpacking.
  95
  96         :param b: Payload from msgpack ext message with code -1
  97         :type b: bytes
  98
  99         :returns: Timestamp object unpacked from msgpack ext payload
 100         :rtype: Timestamp
 101         """
 102         if len(b) == 4:
 103             seconds = struct.unpack("!L", b)[0]
 104             nanoseconds = 0
 105         elif len(b) == 8:
 106             data64 = struct.unpack("!Q", b)[0]
 107             seconds = data64 & 0x00000003FFFFFFFF
 108             nanoseconds = data64 >> 34
 109         elif len(b) == 12:
 110             nanoseconds, seconds = struct.unpack("!Iq", b)
 111         else:
 112             raise ValueError(
 113                 "Timestamp type can only be created from 32, 64, or 96-bit byte objects"
 114             )
 115         return Timestamp(seconds, nanoseconds)
 116
 117     def to_bytes(self):
 118         """Pack this Timestamp object into bytes.
 119
 120         Used for pure-Python msgpack packing.
 121
 122         :returns data: Payload for EXT message with code -1 (timestamp type)
 123         :rtype: bytes
 124         """
 125         if (self.seconds >> 34) == 0:  # seconds is non-negative and fits in 34 bits
 126             data64 = self.nanoseconds << 34 | self.seconds
 127             if data64 & 0xFFFFFFFF00000000 == 0:
 128                 # nanoseconds is zero and seconds < 2**32, so timestamp 32
 129                 data = struct.pack("!L", data64)
 130             else:
 131                 # timestamp 64
 132                 data = struct.pack("!Q", data64)
 133         else:
 134             # timestamp 96
 135             data = struct.pack("!Iq", self.nanoseconds, self.seconds)
 136         return data
 137
 138     @staticmethod
 139     def from_unix(unix_sec):
 140         """Create a Timestamp from posix timestamp in seconds.
 141
 142         :param unix_float: Posix timestamp in seconds.
 143         :type unix_float: int or float.
 144         """
 145         seconds = int(unix_sec // 1)
 146         nanoseconds = int((unix_sec % 1) * 10**9)
 147         return Timestamp(seconds, nanoseconds)
 148
 149     def to_unix(self):
 150         """Get the timestamp as a floating-point value.
 151
 152         :returns: posix timestamp
 153         :rtype: float
 154         """
 155         return self.seconds + self.nanoseconds / 1e9
 156
 157     @staticmethod
 158     def from_unix_nano(unix_ns):
 159         """Create a Timestamp from posix timestamp in nanoseconds.
 160
 161         :param int unix_ns: Posix timestamp in nanoseconds.
 162         :rtype: Timestamp
 163         """
 164         return Timestamp(*divmod(unix_ns, 10**9))
 165
 166     def to_unix_nano(self):
 167         """Get the timestamp as a unixtime in nanoseconds.
 168
 169         :returns: posix timestamp in nanoseconds
 170         :rtype: int
 171         """
 172         return self.seconds * 10**9 + self.nanoseconds
 173
 174     def to_datetime(self):
 175         """Get the timestamp as a UTC datetime.
 176
 177         Python 2 is not supported.
 178
 179         :rtype: datetime.
 180         """
 181         return datetime.datetime.fromtimestamp(0, _utc) + datetime.timedelta(
 182             seconds=self.to_unix()
 183         )
 184
 185     @staticmethod
 186     def from_datetime(dt):
 187         """Create a Timestamp from datetime with tzinfo.
 188
 189         Python 2 is not supported.
 190
 191         :rtype: Timestamp
 192         """
 193         return Timestamp.from_unix(dt.timestamp())