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

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" 

176 

177 def __init__(self, socket_read_size: int): 

178 self._stream: Optional[StreamReader] = None 

179 self._read_size = socket_read_size 

180 

181 @deprecated_function( 

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

183 ) 

184 @abstractmethod 

185 async def can_read_destructive(self) -> bool: 

186 pass 

187 

188 @abstractmethod 

189 async def can_read(self) -> bool: 

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

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

192 pass 

193 

194 async def read_response( 

195 self, disable_decoding: bool = False 

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

197 raise NotImplementedError() 

198 

199 

200class MaintenanceNotificationsParser: 

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

202 

203 @staticmethod 

204 def parse_oss_maintenance_start_msg(response): 

205 # Expected message format is: 

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

207 id = response[1] 

208 slots = safe_str(response[2]) 

209 return OSSNodeMigratingNotification(id, slots) 

210 

211 @staticmethod 

212 def parse_oss_maintenance_completed_msg(response): 

213 # Expected message format is: 

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

215 id = response[1] 

216 nodes_to_slots_mapping_data = response[2] 

217 # Build the nodes_to_slots_mapping dict structure: 

218 # { 

219 # "src_host:port": [ 

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

221 # ... 

222 # ], 

223 # ... 

224 # } 

225 nodes_to_slots_mapping = {} 

226 for src_node, dest_node, slots in nodes_to_slots_mapping_data: 

227 src_node_str = safe_str(src_node) 

228 dest_node_str = safe_str(dest_node) 

229 slots_str = safe_str(slots) 

230 

231 if src_node_str not in nodes_to_slots_mapping: 

232 nodes_to_slots_mapping[src_node_str] = [] 

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

234 

235 return OSSNodeMigratedNotification(id, nodes_to_slots_mapping) 

236 

237 @staticmethod 

238 def parse_maintenance_start_msg(response, notification_type): 

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

240 # Examples: 

241 # MIGRATING 1 10 

242 # FAILING_OVER 2 20 

243 id = response[1] 

244 ttl = response[2] 

245 return notification_type(id, ttl) 

246 

247 @staticmethod 

248 def parse_maintenance_completed_msg(response, notification_type): 

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

250 # Examples: 

251 # MIGRATED 1 

252 # FAILED_OVER 2 

253 id = response[1] 

254 return notification_type(id) 

255 

256 @staticmethod 

257 def parse_moving_msg(response): 

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

259 id = response[1] 

260 ttl = response[2] 

261 if response[3] is None: 

262 host, port = None, None 

263 else: 

264 value = safe_str(response[3]) 

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

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

267 

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

269 

270 

271_INVALIDATION_MESSAGE = "invalidate" 

272_MOVING_MESSAGE = "MOVING" 

273_MIGRATING_MESSAGE = "MIGRATING" 

274_MIGRATED_MESSAGE = "MIGRATED" 

275_FAILING_OVER_MESSAGE = "FAILING_OVER" 

276_FAILED_OVER_MESSAGE = "FAILED_OVER" 

277_SMIGRATING_MESSAGE = "SMIGRATING" 

278_SMIGRATED_MESSAGE = "SMIGRATED" 

279 

280_MAINTENANCE_MESSAGES = ( 

281 _MIGRATING_MESSAGE, 

282 _MIGRATED_MESSAGE, 

283 _FAILING_OVER_MESSAGE, 

284 _FAILED_OVER_MESSAGE, 

285 _SMIGRATING_MESSAGE, 

286) 

287 

288MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING: dict[ 

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

290] = { 

291 _MIGRATING_MESSAGE: ( 

292 NodeMigratingNotification, 

293 MaintenanceNotificationsParser.parse_maintenance_start_msg, 

294 ), 

295 _MIGRATED_MESSAGE: ( 

296 NodeMigratedNotification, 

297 MaintenanceNotificationsParser.parse_maintenance_completed_msg, 

298 ), 

299 _FAILING_OVER_MESSAGE: ( 

300 NodeFailingOverNotification, 

301 MaintenanceNotificationsParser.parse_maintenance_start_msg, 

302 ), 

303 _FAILED_OVER_MESSAGE: ( 

304 NodeFailedOverNotification, 

305 MaintenanceNotificationsParser.parse_maintenance_completed_msg, 

306 ), 

307 _MOVING_MESSAGE: ( 

308 NodeMovingNotification, 

309 MaintenanceNotificationsParser.parse_moving_msg, 

310 ), 

311 _SMIGRATING_MESSAGE: ( 

312 OSSNodeMigratingNotification, 

313 MaintenanceNotificationsParser.parse_oss_maintenance_start_msg, 

314 ), 

315 _SMIGRATED_MESSAGE: ( 

316 OSSNodeMigratedNotification, 

317 MaintenanceNotificationsParser.parse_oss_maintenance_completed_msg, 

318 ), 

319} 

320 

321 

322class PushNotificationsParser(Protocol): 

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

324 

325 pubsub_push_handler_func: Callable 

326 invalidation_push_handler_func: Optional[Callable] = None 

327 node_moving_push_handler_func: Optional[Callable] = None 

328 maintenance_push_handler_func: Optional[Callable] = None 

329 oss_cluster_maint_push_handler_func: Optional[Callable] = None 

330 

331 def handle_pubsub_push_response(self, response): 

332 """Handle pubsub push responses""" 

333 raise NotImplementedError() 

334 

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

336 msg_type = response[0] 

337 if isinstance(msg_type, bytes): 

338 msg_type = msg_type.decode() 

339 

340 if msg_type not in ( 

341 _INVALIDATION_MESSAGE, 

342 *_MAINTENANCE_MESSAGES, 

343 _MOVING_MESSAGE, 

344 _SMIGRATED_MESSAGE, 

345 ): 

346 return self.pubsub_push_handler_func(response) 

347 

348 try: 

349 if ( 

350 msg_type == _INVALIDATION_MESSAGE 

351 and self.invalidation_push_handler_func 

352 ): 

353 return self.invalidation_push_handler_func(response) 

354 

355 if msg_type == _MOVING_MESSAGE and self.node_moving_push_handler_func: 

356 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

357 msg_type 

358 ][1] 

359 

360 notification = parser_function(response) 

361 return self.node_moving_push_handler_func(notification) 

362 

363 if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func: 

364 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

365 msg_type 

366 ][1] 

367 if msg_type == _SMIGRATING_MESSAGE: 

368 notification = parser_function(response) 

369 else: 

370 notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

371 msg_type 

372 ][0] 

373 notification = parser_function(response, notification_type) 

374 

375 if notification is not None: 

376 return self.maintenance_push_handler_func(notification) 

377 if msg_type == _SMIGRATED_MESSAGE and ( 

378 self.oss_cluster_maint_push_handler_func 

379 or self.maintenance_push_handler_func 

380 ): 

381 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

382 msg_type 

383 ][1] 

384 notification = parser_function(response) 

385 

386 if notification is not None: 

387 if self.maintenance_push_handler_func: 

388 self.maintenance_push_handler_func(notification) 

389 if self.oss_cluster_maint_push_handler_func: 

390 self.oss_cluster_maint_push_handler_func(notification) 

391 except Exception as e: 

392 logger.error( 

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

394 ) 

395 

396 return None 

397 

398 def set_pubsub_push_handler(self, pubsub_push_handler_func): 

399 self.pubsub_push_handler_func = pubsub_push_handler_func 

400 

401 def set_invalidation_push_handler(self, invalidation_push_handler_func): 

402 self.invalidation_push_handler_func = invalidation_push_handler_func 

403 

404 def set_node_moving_push_handler(self, node_moving_push_handler_func): 

405 self.node_moving_push_handler_func = node_moving_push_handler_func 

406 

407 def set_maintenance_push_handler(self, maintenance_push_handler_func): 

408 self.maintenance_push_handler_func = maintenance_push_handler_func 

409 

410 def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func): 

411 self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func 

412 

413 

414class AsyncPushNotificationsParser(Protocol): 

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

416 

417 pubsub_push_handler_func: Callable 

418 invalidation_push_handler_func: Optional[Callable] = None 

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

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

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

422 

423 async def handle_pubsub_push_response(self, response): 

424 """Handle pubsub push responses asynchronously""" 

425 raise NotImplementedError() 

426 

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

428 """Handle push responses asynchronously""" 

429 

430 msg_type = response[0] 

431 if isinstance(msg_type, bytes): 

432 msg_type = msg_type.decode() 

433 

434 if msg_type not in ( 

435 _INVALIDATION_MESSAGE, 

436 *_MAINTENANCE_MESSAGES, 

437 _MOVING_MESSAGE, 

438 _SMIGRATED_MESSAGE, 

439 ): 

440 return await self.pubsub_push_handler_func(response) 

441 

442 try: 

443 if ( 

444 msg_type == _INVALIDATION_MESSAGE 

445 and self.invalidation_push_handler_func 

446 ): 

447 return await self.invalidation_push_handler_func(response) 

448 

449 if isinstance(msg_type, bytes): 

450 msg_type = msg_type.decode() 

451 

452 if msg_type == _MOVING_MESSAGE and self.node_moving_push_handler_func: 

453 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

454 msg_type 

455 ][1] 

456 notification = parser_function(response) 

457 return await self.node_moving_push_handler_func(notification) 

458 

459 if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func: 

460 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

461 msg_type 

462 ][1] 

463 if msg_type == _SMIGRATING_MESSAGE: 

464 notification = parser_function(response) 

465 else: 

466 notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

467 msg_type 

468 ][0] 

469 notification = parser_function(response, notification_type) 

470 

471 if notification is not None: 

472 return await self.maintenance_push_handler_func(notification) 

473 if ( 

474 msg_type == _SMIGRATED_MESSAGE 

475 and self.oss_cluster_maint_push_handler_func 

476 ): 

477 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[ 

478 msg_type 

479 ][1] 

480 notification = parser_function(response) 

481 if notification is not None: 

482 return await self.oss_cluster_maint_push_handler_func(notification) 

483 except Exception as e: 

484 logger.error( 

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

486 ) 

487 

488 return None 

489 

490 def set_pubsub_push_handler(self, pubsub_push_handler_func): 

491 """Set the pubsub push handler function""" 

492 self.pubsub_push_handler_func = pubsub_push_handler_func 

493 

494 def set_invalidation_push_handler(self, invalidation_push_handler_func): 

495 """Set the invalidation push handler function""" 

496 self.invalidation_push_handler_func = invalidation_push_handler_func 

497 

498 def set_node_moving_push_handler(self, node_moving_push_handler_func): 

499 self.node_moving_push_handler_func = node_moving_push_handler_func 

500 

501 def set_maintenance_push_handler(self, maintenance_push_handler_func): 

502 self.maintenance_push_handler_func = maintenance_push_handler_func 

503 

504 def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func): 

505 self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func 

506 

507 

508class _AsyncRESPBase(AsyncBaseParser): 

509 """Base class for async resp parsing""" 

510 

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

512 

513 def __init__(self, socket_read_size: int): 

514 super().__init__(socket_read_size) 

515 self.encoder: Optional[Encoder] = None 

516 self._buffer = b"" 

517 self._chunks = [] 

518 self._pos = 0 

519 

520 def _clear(self): 

521 self._buffer = b"" 

522 self._chunks.clear() 

523 

524 def on_connect(self, connection): 

525 """Called when the stream connects""" 

526 self._stream = connection._reader 

527 if self._stream is None: 

528 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

529 self.encoder = connection.encoder 

530 self._clear() 

531 self._connected = True 

532 

533 def on_disconnect(self): 

534 """Called when the stream disconnects""" 

535 self._connected = False 

536 

537 @deprecated_function( 

538 version="8.0.0", 

539 reason="Use can_read() instead", 

540 name="can_read_destructive", 

541 ) 

542 async def can_read_destructive(self) -> bool: 

543 return await self.can_read() 

544 

545 async def can_read(self) -> bool: 

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

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

548 if not self._connected: 

549 raise OSError("Buffer is closed.") 

550 if self._buffer: 

551 return True 

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

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

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

555 return bool(self._stream._buffer) or self._stream.at_eof() 

556 

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

558 """ 

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

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

561 """ 

562 want = length + 2 

563 end = self._pos + want 

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

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

566 else: 

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

568 try: 

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

570 except IncompleteReadError as error: 

571 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error 

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

573 self._chunks.append(data) 

574 self._pos += want 

575 return result 

576 

577 async def _readline(self) -> bytes: 

578 """ 

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

580 line separator, which is discarded. 

581 """ 

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

583 if found >= 0: 

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

585 else: 

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

587 data = await self._stream.readline() 

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

589 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) 

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

591 self._chunks.append(data) 

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

593 return result