Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/_parsers/base.py: 35%

1import logging 

2from abc import ABC, abstractmethod 

3from asyncio import IncompleteReadError, StreamReader 

4from typing import Awaitable, Callable, List, Optional, Protocol, Union 

5 

6from redis.maint_notifications import ( 

7 MaintenanceNotification, 

8 NodeFailedOverNotification, 

9 NodeFailingOverNotification, 

10 NodeMigratedNotification, 

11 NodeMigratingNotification, 

12 NodeMovingNotification, 

13 OSSNodeMigratedNotification, 

14 OSSNodeMigratingNotification, 

15) 

16from redis.utils import deprecated_function, safe_str 

17 

18from ..exceptions import ( 

19 AskError, 

20 AuthenticationError, 

21 AuthenticationWrongNumberOfArgsError, 

22 BusyLoadingError, 

23 ClusterCrossSlotError, 

24 ClusterDownError, 

25 ConnectionError, 

26 ExecAbortError, 

27 ExternalAuthProviderError, 

28 MasterDownError, 

29 ModuleError, 

30 MovedError, 

31 NoPermissionError, 

32 NoScriptError, 

33 NoSuchFieldsetError, 

34 OutOfMemoryError, 

35 ReadOnlyError, 

36 ResponseError, 

37 TryAgainError, 

38) 

39from ..typing import EncodableT 

40from .encoders import Encoder 

41from .socket import SERVER_CLOSED_CONNECTION_ERROR, SocketBuffer 

42 

43MODULE_LOAD_ERROR = "Error loading the extension. Please check the server logs." 

44NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name" 

45MODULE_UNLOAD_NOT_POSSIBLE_ERROR = "Error unloading module: operation not possible." 

46MODULE_EXPORTS_DATA_TYPES_ERROR = ( 

47 "Error unloading module: the module " 

48 "exports one or more module-side data " 

49 "types, can't unload" 

50) 

51# user send an AUTH cmd to a server without authorization configured 

52NO_AUTH_SET_ERROR = { 

53 # Redis >= 6.0 

54 "AUTH <password> called without any password " 

55 "configured for the default user. Are you sure " 

56 "your configuration is correct?": AuthenticationError, 

57 # Redis < 6.0 

58 "Client sent AUTH, but no password is set": AuthenticationError, 

59} 

60 

61EXTERNAL_AUTH_PROVIDER_ERROR = { 

62 "problem with LDAP service": ExternalAuthProviderError, 

63} 

64 

65# HIMPORT SET referencing a fieldset the connection has not prepared. The server 

66# reply is a fixed message with no fieldset name appended (verified against the 

67# server: always exactly ``ERR no such fieldset``), so an exact match is correct. 

68NO_SUCH_FIELDSET_ERROR = { 

69 "no such fieldset": NoSuchFieldsetError, 

70} 

71 

72logger = logging.getLogger(__name__) 

73 

74 

75class BaseParser(ABC): 

76 EXCEPTION_CLASSES = { 

77 "ERR": { 

78 "max number of clients reached": ConnectionError, 

79 "invalid password": AuthenticationError, 

80 # some Redis server versions report invalid command syntax 

81 # in lowercase 

82 "wrong number of arguments " 

83 "for 'auth' command": AuthenticationWrongNumberOfArgsError, 

84 # some Redis server versions report invalid command syntax 

85 # in uppercase 

86 "wrong number of arguments " 

87 "for 'AUTH' command": AuthenticationWrongNumberOfArgsError, 

88 MODULE_LOAD_ERROR: ModuleError, 

89 MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError, 

90 NO_SUCH_MODULE_ERROR: ModuleError, 

91 MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError, 

92 **NO_AUTH_SET_ERROR, 

93 **EXTERNAL_AUTH_PROVIDER_ERROR, 

94 **NO_SUCH_FIELDSET_ERROR, 

95 }, 

96 "OOM": OutOfMemoryError, 

97 "WRONGPASS": AuthenticationError, 

98 "EXECABORT": ExecAbortError, 

99 "LOADING": BusyLoadingError, 

100 "NOSCRIPT": NoScriptError, 

101 "READONLY": ReadOnlyError, 

102 "NOAUTH": AuthenticationError, 

103 "NOPERM": NoPermissionError, 

104 "ASK": AskError, 

105 "TRYAGAIN": TryAgainError, 

106 "MOVED": MovedError, 

107 "CLUSTERDOWN": ClusterDownError, 

108 "CROSSSLOT": ClusterCrossSlotError, 

109 "MASTERDOWN": MasterDownError, 

110 } 

111 

112 @classmethod 

113 def parse_error(cls, response): 

114 "Parse an error response" 

115 error_code = response.split(" ")[0] 

116 if error_code in cls.EXCEPTION_CLASSES: 

117 response = response[len(error_code) + 1 :] 

118 exception_class = cls.EXCEPTION_CLASSES[error_code] 

119 if isinstance(exception_class, dict): 

120 exception_class = exception_class.get(response, ResponseError) 

121 return exception_class(response, status_code=error_code) 

122 return ResponseError(response) 

123 

124 @abstractmethod 

125 def on_disconnect(self): 

126 pass 

127 

128 @abstractmethod 

129 def on_connect(self, connection): 

130 pass 

131 

132 

133class _RESPBase(BaseParser): 

134 """Base class for sync-based resp parsing""" 

135 

136 def __init__(self, socket_read_size): 

137 self.socket_read_size = socket_read_size 

138 self.encoder = None 

139 self._sock = None 

140 self._buffer = None 

141 

142 def __del__(self): 

143 try: 

144 self.on_disconnect() 

145 except Exception: 

146 pass 

147 

148 def on_connect(self, connection): 

149 "Called when the socket connects" 

150 self._sock = connection._sock 

151 self._buffer = SocketBuffer( 

152 self._sock, self.socket_read_size, connection.socket_timeout 

153 ) 

154 self.encoder = connection.encoder 

155 

156 def on_disconnect(self): 

157 "Called when the socket disconnects" 

158 self._sock = None 

159 if self._buffer is not None: 

160 self._buffer.close() 

161 self._buffer = None 

162 self.encoder = None 

163 

164 def can_read(self, timeout: float = 0) -> bool: 

165 # TODO: Rename this API; it detects pending data or dirty/closed 

166 # connection state, not only whether application data can be read. 

167 if self._buffer is None: 

168 return False 

169 return self._buffer.can_read(timeout) 

170 

171 

172class AsyncBaseParser(BaseParser): 

173 """Base parsing class for the python-backed async parser""" 

174 

175 __slots__ = "_stream", "_read_size", "_connected" 

176 

177 def __init__(self, socket_read_size: int): 

178 self._stream: Optional[StreamReader] = None 

179 self._read_size = socket_read_size 

180 # Initialized here, not only in on_connect() / on_disconnect(): the 

181 # read guards in can_read() and the subclasses' read_response() consult 

182 # this flag to turn a read on a dead connection into a retryable 

183 # ConnectionError. A parser that has never connected would otherwise 

184 # raise AttributeError from those guards, which no retry layer acts on. 

185 self._connected: bool = False 

186 

187 @deprecated_function( 

188 version="8.0.0", reason="Use can_read() instead", name="can_read_destructive" 

189 ) 

190 @abstractmethod 

191 async def can_read_destructive(self) -> bool: 

192 pass 

193 

194 @abstractmethod 

195 async def can_read(self) -> bool: 

196 # TODO: Rename this API; it detects pending data or dirty/closed 

197 # connection state, not only whether application data can be read. 

198 pass 

199 

200 async def read_response( 

201 self, disable_decoding: bool = False 

202 ) -> Union[EncodableT, ResponseError, None, List[EncodableT]]: 

203 raise NotImplementedError() 

204 

205 

206class MaintenanceNotificationsParser: 

207 """Protocol defining maintenance push notification parsing functionality""" 

208 

209 @staticmethod 

210 def parse_oss_maintenance_start_msg(response): 

211 # Expected message format is: 

212 # SMIGRATING <seq_number> <slot, range1-range2,...> 

213 id = response[1] 

214 slots = safe_str(response[2]) 

215 return OSSNodeMigratingNotification(id, slots) 

216 

217 @staticmethod 

218 def parse_oss_maintenance_completed_msg(response): 

219 # Expected message format is: 

220 # SMIGRATED <seq_number> [[<src_host:port> <dest_host:port> <slot_range>], ...] 

221 id = response[1] 

222 nodes_to_slots_mapping_data = response[2] 

223 # Build the nodes_to_slots_mapping dict structure: 

224 # { 

225 # "src_host:port": [ 

226 # {"dest_host:port": "slot_range"}, 

227 # ... 

228 # ], 

229 # ... 

230 # } 

231 nodes_to_slots_mapping = {} 

232 for src_node, dest_node, slots in nodes_to_slots_mapping_data: 

233 src_node_str = safe_str(src_node) 

234 dest_node_str = safe_str(dest_node) 

235 slots_str = safe_str(slots) 

236 

237 if src_node_str not in nodes_to_slots_mapping: 

238 nodes_to_slots_mapping[src_node_str] = [] 

239 nodes_to_slots_mapping[src_node_str].append({dest_node_str: slots_str}) 

240 

241 return OSSNodeMigratedNotification(id, nodes_to_slots_mapping) 

242 

243 @staticmethod 

244 def parse_maintenance_start_msg(response, notification_type): 

245 # Expected message format is: <notification_type> <seq_number> <time> 

246 # Examples: 

247 # MIGRATING 1 10 

248 # FAILING_OVER 2 20 

249 id = response[1] 

250 ttl = response[2] 

251 return notification_type(id, ttl) 

252 

253 @staticmethod 

254 def parse_maintenance_completed_msg(response, notification_type): 

255 # Expected message format is: <notification_type> <seq_number> 

256 # Examples: 

257 # MIGRATED 1 

258 # FAILED_OVER 2 

259 id = response[1] 

260 return notification_type(id) 

261 

262 @staticmethod 

263 def parse_moving_msg(response): 

264 # Expected message format is: MOVING <seq_number> <time> <endpoint> 

265 id = response[1] 

266 ttl = response[2] 

267 if response[3] is None: 

268 host, port = None, None 

269 else: 

270 value = safe_str(response[3]) 

271 host, port = value.rsplit(":", 1) 

272 port = int(port) if port is not None else None 

273 

274 return NodeMovingNotification(id, host, port, ttl) 

275 

276 

277_INVALIDATION_MESSAGE = "invalidate" 

278_MOVING_MESSAGE = "MOVING" 

279_MIGRATING_MESSAGE = "MIGRATING" 

280_MIGRATED_MESSAGE = "MIGRATED" 

281_FAILING_OVER_MESSAGE = "FAILING_OVER" 

282_FAILED_OVER_MESSAGE = "FAILED_OVER" 

283_SMIGRATING_MESSAGE = "SMIGRATING" 

284_SMIGRATED_MESSAGE = "SMIGRATED" 

285 

286_MAINTENANCE_MESSAGES = ( 

287 _MIGRATING_MESSAGE, 

288 _MIGRATED_MESSAGE, 

289 _FAILING_OVER_MESSAGE, 

290 _FAILED_OVER_MESSAGE, 

291 _SMIGRATING_MESSAGE, 

292) 

293 

294MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING: dict[ 

295 str, tuple[type[MaintenanceNotification], Callable] 

296] = { 

297 _MIGRATING_MESSAGE: ( 

298 NodeMigratingNotification, 

299 MaintenanceNotificationsParser.parse_maintenance_start_msg, 

300 ), 

301 _MIGRATED_MESSAGE: ( 

302 NodeMigratedNotification, 

303 MaintenanceNotificationsParser.parse_maintenance_completed_msg, 

304 ), 

305 _FAILING_OVER_MESSAGE: ( 

306 NodeFailingOverNotification, 

307 MaintenanceNotificationsParser.parse_maintenance_start_msg, 

308 ), 

309 _FAILED_OVER_MESSAGE: ( 

310 NodeFailedOverNotification, 

311 MaintenanceNotificationsParser.parse_maintenance_completed_msg, 

312 ), 

313 _MOVING_MESSAGE: ( 

314 NodeMovingNotification, 

315 MaintenanceNotificationsParser.parse_moving_msg, 

316 ), 

317 _SMIGRATING_MESSAGE: ( 

318 OSSNodeMigratingNotification, 

319 MaintenanceNotificationsParser.parse_oss_maintenance_start_msg, 

320 ), 

321 _SMIGRATED_MESSAGE: ( 

322 OSSNodeMigratedNotification, 

323 MaintenanceNotificationsParser.parse_oss_maintenance_completed_msg, 

324 ), 

325} 

326 

327 

328class PushNotificationsParser(Protocol): 

329 """Protocol defining RESP3-specific parsing functionality""" 

330 

331 pubsub_push_handler_func: Callable 

332 invalidation_push_handler_func: Optional[Callable] = None 

333 node_moving_push_handler_func: Optional[Callable] = None 

334 maintenance_push_handler_func: Optional[Callable] = None 

335 oss_cluster_maint_push_handler_func: Optional[Callable] = None 

336 

337 def handle_pubsub_push_response(self, response): 

338 """Handle pubsub push responses""" 

339 raise NotImplementedError() 

340 

341 def handle_push_response(self, response, **kwargs): 

342 msg_type = response[0] 

343 if isinstance(msg_type, bytes): 

344 msg_type = msg_type.decode() 

345 

346 if msg_type not in ( 

347 _INVALIDATION_MESSAGE, 

348 *_MAINTENANCE_MESSAGES, 

349 _MOVING_MESSAGE, 

350 _SMIGRATED_MESSAGE, 

351 ): 

352 return self.pubsub_push_handler_func(response) 

353 

354 try: 

355 if ( 

356 msg_type == _INVALIDATION_MESSAGE 

357 and self.invalidation_push_handler_func 

358 ): 

359 return self.invalidation_push_handler_func(response) 

360 

361 if msg_type == _MOVING_MESSAGE and self.node_moving_push_handler_func: 

362 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

363 msg_type 

364 ][1] 

365 

366 notification = parser_function(response) 

367 return self.node_moving_push_handler_func(notification) 

368 

369 if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func: 

370 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

371 msg_type 

372 ][1] 

373 if msg_type == _SMIGRATING_MESSAGE: 

374 notification = parser_function(response) 

375 else: 

376 notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

377 msg_type 

378 ][0] 

379 notification = parser_function(response, notification_type) 

380 

381 if notification is not None: 

382 return self.maintenance_push_handler_func(notification) 

383 if msg_type == _SMIGRATED_MESSAGE and ( 

384 self.oss_cluster_maint_push_handler_func 

385 or self.maintenance_push_handler_func 

386 ): 

387 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

388 msg_type 

389 ][1] 

390 notification = parser_function(response) 

391 

392 if notification is not None: 

393 if self.maintenance_push_handler_func: 

394 self.maintenance_push_handler_func(notification) 

395 if self.oss_cluster_maint_push_handler_func: 

396 self.oss_cluster_maint_push_handler_func(notification) 

397 except Exception as e: 

398 logger.error( 

399 "Error handling {} message ({}): {}".format(msg_type, response, e) 

400 ) 

401 

402 return None 

403 

404 def set_pubsub_push_handler(self, pubsub_push_handler_func): 

405 self.pubsub_push_handler_func = pubsub_push_handler_func 

406 

407 def set_invalidation_push_handler(self, invalidation_push_handler_func): 

408 self.invalidation_push_handler_func = invalidation_push_handler_func 

409 

410 def set_node_moving_push_handler(self, node_moving_push_handler_func): 

411 self.node_moving_push_handler_func = node_moving_push_handler_func 

412 

413 def set_maintenance_push_handler(self, maintenance_push_handler_func): 

414 self.maintenance_push_handler_func = maintenance_push_handler_func 

415 

416 def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func): 

417 self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func 

418 

419 

420class AsyncPushNotificationsParser(Protocol): 

421 """Protocol defining async RESP3-specific parsing functionality""" 

422 

423 pubsub_push_handler_func: Callable 

424 invalidation_push_handler_func: Optional[Callable] = None 

425 node_moving_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None 

426 maintenance_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None 

427 oss_cluster_maint_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None 

428 

429 async def handle_pubsub_push_response(self, response): 

430 """Handle pubsub push responses asynchronously""" 

431 raise NotImplementedError() 

432 

433 async def handle_push_response(self, response, **kwargs): 

434 """Handle push responses asynchronously""" 

435 

436 msg_type = response[0] 

437 if isinstance(msg_type, bytes): 

438 msg_type = msg_type.decode() 

439 

440 if msg_type not in ( 

441 _INVALIDATION_MESSAGE, 

442 *_MAINTENANCE_MESSAGES, 

443 _MOVING_MESSAGE, 

444 _SMIGRATED_MESSAGE, 

445 ): 

446 return await self.pubsub_push_handler_func(response) 

447 

448 try: 

449 if ( 

450 msg_type == _INVALIDATION_MESSAGE 

451 and self.invalidation_push_handler_func 

452 ): 

453 return await self.invalidation_push_handler_func(response) 

454 

455 if isinstance(msg_type, bytes): 

456 msg_type = msg_type.decode() 

457 

458 if msg_type == _MOVING_MESSAGE and self.node_moving_push_handler_func: 

459 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

460 msg_type 

461 ][1] 

462 notification = parser_function(response) 

463 return await self.node_moving_push_handler_func(notification) 

464 

465 if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func: 

466 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

467 msg_type 

468 ][1] 

469 if msg_type == _SMIGRATING_MESSAGE: 

470 notification = parser_function(response) 

471 else: 

472 notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

473 msg_type 

474 ][0] 

475 notification = parser_function(response, notification_type) 

476 

477 if notification is not None: 

478 return await self.maintenance_push_handler_func(notification) 

479 if ( 

480 msg_type == _SMIGRATED_MESSAGE 

481 and self.oss_cluster_maint_push_handler_func 

482 ): 

483 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

484 msg_type 

485 ][1] 

486 notification = parser_function(response) 

487 if notification is not None: 

488 return await self.oss_cluster_maint_push_handler_func(notification) 

489 except Exception as e: 

490 logger.error( 

491 "Error handling {} message ({}): {}".format(msg_type, response, e) 

492 ) 

493 

494 return None 

495 

496 def set_pubsub_push_handler(self, pubsub_push_handler_func): 

497 """Set the pubsub push handler function""" 

498 self.pubsub_push_handler_func = pubsub_push_handler_func 

499 

500 def set_invalidation_push_handler(self, invalidation_push_handler_func): 

501 """Set the invalidation push handler function""" 

502 self.invalidation_push_handler_func = invalidation_push_handler_func 

503 

504 def set_node_moving_push_handler(self, node_moving_push_handler_func): 

505 self.node_moving_push_handler_func = node_moving_push_handler_func 

506 

507 def set_maintenance_push_handler(self, maintenance_push_handler_func): 

508 self.maintenance_push_handler_func = maintenance_push_handler_func 

509 

510 def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func): 

511 self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func 

512 

513 

514class _AsyncRESPBase(AsyncBaseParser): 

515 """Base class for async resp parsing""" 

516 

517 __slots__ = AsyncBaseParser.__slots__ + ("encoder", "_buffer", "_pos", "_chunks") 

518 

519 def __init__(self, socket_read_size: int): 

520 super().__init__(socket_read_size) 

521 self.encoder: Optional[Encoder] = None 

522 self._buffer = b"" 

523 self._chunks = [] 

524 self._pos = 0 

525 

526 def _clear(self): 

527 self._buffer = b"" 

528 self._chunks.clear() 

529 

530 def on_connect(self, connection): 

531 """Called when the stream connects""" 

532 self._stream = connection._reader 

533 if self._stream is None: 

534 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

535 self.encoder = connection.encoder 

536 self._clear() 

537 self._connected = True 

538 

539 def on_disconnect(self): 

540 """Called when the stream disconnects""" 

541 self._connected = False 

542 

543 @deprecated_function( 

544 version="8.0.0", 

545 reason="Use can_read() instead", 

546 name="can_read_destructive", 

547 ) 

548 async def can_read_destructive(self) -> bool: 

549 return await self.can_read() 

550 

551 async def can_read(self) -> bool: 

552 # TODO: Rename this API; it detects pending data or dirty/closed 

553 # connection state, not only whether application data can be read. 

554 if not self._connected: 

555 raise OSError("Buffer is closed.") 

556 # buffered data wins over EOF, like the sync SocketBuffer: a pending 

557 # response or push notification must stay readable even if the server 

558 # has since closed the connection. 

559 if self._buffer: 

560 return True 

561 if self._stream.at_eof(): 

562 # Raise like the sync SocketBuffer does on a server-closed 

563 # connection, so callers that tolerate pending data (push 

564 # notifications) can't mistake EOF for a readable connection. 

565 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

566 # asyncio.StreamReader has no public non-destructive API for checking 

567 # buffered bytes. Preserve dirty-connection detection for the Python 

568 # parser and fail loudly if the private buffer API changes. 

569 return bool(self._stream._buffer) 

570 

571 async def _read(self, length: int) -> bytes: 

572 """ 

573 Read `length` bytes of data. These are assumed to be followed 

574 by a '\r\n' terminator which is subsequently discarded. 

575 """ 

576 want = length + 2 

577 end = self._pos + want 

578 if len(self._buffer) >= end: 

579 result = self._buffer[self._pos : end - 2] 

580 else: 

581 tail = self._buffer[self._pos :] 

582 try: 

583 data = await self._stream.readexactly(want - len(tail)) 

584 except IncompleteReadError as error: 

585 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error 

586 result = (tail + data)[:-2] 

587 self._chunks.append(data) 

588 self._pos += want 

589 return result 

590 

591 async def _readline(self) -> bytes: 

592 """ 

593 read an unknown number of bytes up to the next '\r\n' 

594 line separator, which is discarded. 

595 """ 

596 found = self._buffer.find(b"\r\n", self._pos) 

597 if found >= 0: 

598 result = self._buffer[self._pos : found] 

599 else: 

600 tail = self._buffer[self._pos :] 

601 data = await self._stream.readline() 

602 if not data.endswith(b"\r\n"): 

603 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

604 result = (tail + data)[:-2] 

605 self._chunks.append(data) 

606 self._pos += len(result) + 2 

607 return result