Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/msgpack/ext.py: 48%

1import datetime 

2import struct 

3from collections import namedtuple 

4 

5_EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc) 

6 

7 

8class ExtType(namedtuple("ExtType", "code data")): 

9 """ExtType represents ext type in msgpack.""" 

10 

11 def __new__(cls, code, data): 

12 if not isinstance(code, int): 

13 raise TypeError("code must be int") 

14 if not isinstance(data, bytes): 

15 raise TypeError("data must be bytes") 

16 if not 0 <= code <= 127: 

17 raise ValueError("code must be 0~127") 

18 return super().__new__(cls, code, data) 

19 

20 

21class Timestamp: 

22 """Timestamp represents the Timestamp extension type in msgpack. 

23 

24 When built with Cython, msgpack uses C methods to pack and unpack `Timestamp`. 

25 When using pure-Python msgpack, :func:`to_bytes` and :func:`from_bytes` are used to pack and 

26 unpack `Timestamp`. 

27 

28 This class is immutable: Do not override seconds and nanoseconds. 

29 """ 

30 

31 __slots__ = ["seconds", "nanoseconds"] 

32 

33 def __init__(self, seconds, nanoseconds=0): 

34 """Initialize a Timestamp object. 

35 

36 :param int seconds: 

37 Number of seconds since the UNIX epoch (00:00:00 UTC Jan 1 1970, minus leap seconds). 

38 May be negative. 

39 

40 :param int nanoseconds: 

41 Number of nanoseconds to add to `seconds` to get fractional time. 

42 Maximum is 999_999_999. Default is 0. 

43 

44 Note: Negative times (before the UNIX epoch) are represented as neg. seconds + pos. ns. 

45 """ 

46 if not isinstance(seconds, int): 

47 raise TypeError("seconds must be an integer") 

48 if not isinstance(nanoseconds, int): 

49 raise TypeError("nanoseconds must be an integer") 

50 if not (0 <= nanoseconds < 10**9): 

51 raise ValueError("nanoseconds must be a non-negative integer less than 999999999.") 

52 self.seconds = seconds 

53 self.nanoseconds = nanoseconds 

54 

55 def __repr__(self): 

56 """String representation of Timestamp.""" 

57 return f"Timestamp(seconds={self.seconds}, nanoseconds={self.nanoseconds})" 

58 

59 def __eq__(self, other): 

60 """Check for equality with another Timestamp object""" 

61 if type(other) is self.__class__: 

62 return self.seconds == other.seconds and self.nanoseconds == other.nanoseconds 

63 return False 

64 

65 def __ne__(self, other): 

66 """not-equals method (see :func:`__eq__()`)""" 

67 return not self.__eq__(other) 

68 

69 def __hash__(self): 

70 return hash((self.seconds, self.nanoseconds)) 

71 

72 @staticmethod 

73 def from_bytes(b): 

74 """Unpack bytes into a `Timestamp` object. 

75 

76 Used for pure-Python msgpack unpacking. 

77 

78 :param b: Payload from msgpack ext message with code -1 

79 :type b: bytes 

80 

81 :returns: Timestamp object unpacked from msgpack ext payload 

82 :rtype: Timestamp 

83 """ 

84 if len(b) == 4: 

85 seconds = struct.unpack("!L", b)[0] 

86 nanoseconds = 0 

87 elif len(b) == 8: 

88 data64 = struct.unpack("!Q", b)[0] 

89 seconds = data64 & 0x00000003FFFFFFFF 

90 nanoseconds = data64 >> 34 

91 elif len(b) == 12: 

92 nanoseconds, seconds = struct.unpack("!Iq", b) 

93 else: 

94 raise ValueError( 

95 "Timestamp type can only be created from 32, 64, or 96-bit byte objects" 

96 ) 

97 return Timestamp(seconds, nanoseconds) 

98 

99 def to_bytes(self): 

100 """Pack this Timestamp object into bytes. 

101 

102 Used for pure-Python msgpack packing. 

103 

104 :returns data: Payload for EXT message with code -1 (timestamp type) 

105 :rtype: bytes 

106 """ 

107 if (self.seconds >> 34) == 0: # seconds is non-negative and fits in 34 bits 

108 data64 = self.nanoseconds << 34 | self.seconds 

109 if data64 & 0xFFFFFFFF00000000 == 0: 

110 # nanoseconds is zero and seconds < 2**32, so timestamp 32 

111 data = struct.pack("!L", data64) 

112 else: 

113 # timestamp 64 

114 data = struct.pack("!Q", data64) 

115 else: 

116 # timestamp 96 

117 data = struct.pack("!Iq", self.nanoseconds, self.seconds) 

118 return data 

119 

120 @staticmethod 

121 def from_unix(unix_sec): 

122 """Create a Timestamp from posix timestamp in seconds. 

123 

124 :param unix_float: Posix timestamp in seconds. 

125 :type unix_float: int or float 

126 """ 

127 seconds = int(unix_sec // 1) 

128 nanoseconds = int((unix_sec % 1) * 10**9) 

129 return Timestamp(seconds, nanoseconds) 

130 

131 def to_unix(self): 

132 """Get the timestamp as a floating-point value. 

133 

134 :returns: posix timestamp 

135 :rtype: float 

136 """ 

137 return self.seconds + self.nanoseconds / 1e9 

138 

139 @staticmethod 

140 def from_unix_nano(unix_ns): 

141 """Create a Timestamp from posix timestamp in nanoseconds. 

142 

143 :param int unix_ns: Posix timestamp in nanoseconds. 

144 :rtype: Timestamp 

145 """ 

146 return Timestamp(*divmod(unix_ns, 10**9)) 

147 

148 def to_unix_nano(self): 

149 """Get the timestamp as a unixtime in nanoseconds. 

150 

151 :returns: posix timestamp in nanoseconds 

152 :rtype: int 

153 """ 

154 return self.seconds * 10**9 + self.nanoseconds 

155 

156 def to_datetime(self): 

157 """Get the timestamp as a UTC datetime. 

158 

159 :rtype: `datetime.datetime` 

160 """ 

161 return _EPOCH + datetime.timedelta( 

162 seconds=self.seconds, microseconds=self.nanoseconds // 1000 

163 ) 

164 

165 @staticmethod 

166 def from_datetime(dt): 

167 """Create a Timestamp from datetime with tzinfo. 

168 

169 :rtype: Timestamp 

170 """ 

171 # Use integer timedelta arithmetic (like the Cython packer) rather than 

172 # ``int(dt.timestamp() // 1)``. ``datetime.timestamp()`` returns a float 

173 # that cannot hold microsecond precision for datetimes far from the epoch, 

174 # so it may round the whole-second part up while the exact ``microsecond`` 

175 # is still used for the nanoseconds -- producing a Timestamp one second in 

176 # the future (and OverflowError near datetime.max). 

177 if dt.tzinfo is None: 

178 # Match datetime.timestamp(): a naive datetime is treated as local time. 

179 dt = dt.astimezone() 

180 delta = dt - _EPOCH 

181 return Timestamp( 

182 seconds=delta.days * 86400 + delta.seconds, 

183 nanoseconds=delta.microseconds * 1000, 

184 )