Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/grpc/aio/_call.py: 35%

1# Copyright 2019 gRPC authors. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14"""Invocation-side implementation of gRPC Asyncio Python.""" 

15 

16import asyncio 

17from collections.abc import AsyncIterable 

18import enum 

19from functools import partial 

20import logging 

21import traceback 

22from typing import ( 

23 Any, 

24 AsyncIterator, 

25 Generator, 

26 Generic, 

27 Optional, 

28 Tuple, 

29 Union, 

30) 

31 

32import grpc 

33from grpc import _common 

34from grpc._cython import cygrpc 

35 

36from . import _base_call # pyright: ignore[reportPrivateUsage] 

37from ._metadata import Metadata 

38from ._typing import DeserializingFunction 

39from ._typing import DoneCallbackType 

40from ._typing import EOFType 

41from ._typing import MetadataType 

42from ._typing import MetadatumType 

43from ._typing import RequestIterableType 

44from ._typing import RequestType 

45from ._typing import ResponseType 

46from ._typing import SerializingFunction 

47 

48__all__ = "AioRpcError", "Call", "UnaryStreamCall", "UnaryUnaryCall" 

49 

50_LOCAL_CANCELLATION_DETAILS = "Locally cancelled by application!" 

51_GC_CANCELLATION_DETAILS = "Cancelled upon garbage collection!" 

52_RPC_ALREADY_FINISHED_DETAILS = "RPC already finished." 

53_RPC_HALF_CLOSED_DETAILS = 'RPC is half closed after calling "done_writing".' 

54_API_STYLE_ERROR = ( 

55 "The iterator and read/write APIs may not be mixed on a single RPC." 

56) 

57 

58_OK_CALL_REPRESENTATION = ( 

59 '<{} of RPC that terminated with:\n\tstatus = {}\n\tdetails = "{}"\n>' 

60) 

61 

62_NON_OK_CALL_REPRESENTATION = ( 

63 "<{} of RPC that terminated with:\n" 

64 "\tstatus = {}\n" 

65 '\tdetails = "{}"\n' 

66 '\tdebug_error_string = "{}"\n' 

67 ">" 

68) 

69 

70_LOGGER = logging.getLogger(__name__) 

71 

72 

73class AioRpcError(grpc.RpcError): 

74 """An implementation of RpcError to be used by the asynchronous API. 

75 

76 Raised RpcError is a snapshot of the final status of the RPC, values are 

77 determined. Hence, its methods no longer needs to be coroutines. 

78 """ 

79 

80 _code: grpc.StatusCode 

81 _details: Optional[str] 

82 _initial_metadata: Optional[Metadata] = None 

83 _trailing_metadata: Optional[Metadata] = None 

84 _debug_error_string: Optional[str] 

85 

86 def __init__( 

87 self, 

88 code: grpc.StatusCode, 

89 initial_metadata: Optional[Metadata] = None, 

90 trailing_metadata: Optional[Metadata] = None, 

91 details: Optional[str] = None, 

92 debug_error_string: Optional[str] = None, 

93 ) -> None: 

94 """Constructor. 

95 

96 Args: 

97 code: The status code with which the RPC has been finalized. 

98 initial_metadata: Optional initial metadata that could be sent by the 

99 Server. 

100 trailing_metadata: Optional metadata that could be sent by the Server. 

101 details: Optional details explaining the reason of the error. 

102 debug_error_string: Optional string 

103 """ 

104 super().__init__() 

105 self._code = code 

106 self._details = details 

107 self._initial_metadata = initial_metadata 

108 self._trailing_metadata = trailing_metadata 

109 self._debug_error_string = debug_error_string 

110 

111 def code(self) -> grpc.StatusCode: 

112 """Accesses the status code sent by the server. 

113 

114 Returns: 

115 The `grpc.StatusCode` status code. 

116 """ 

117 return self._code 

118 

119 def details(self) -> Optional[str]: 

120 """Accesses the details sent by the server. 

121 

122 Returns: 

123 The description of the error. 

124 """ 

125 return self._details 

126 

127 def initial_metadata(self) -> Optional[Metadata]: 

128 """Accesses the initial metadata sent by the server. 

129 

130 Returns: 

131 The initial metadata received. 

132 """ 

133 return self._initial_metadata 

134 

135 def trailing_metadata(self) -> Optional[Metadata]: 

136 """Accesses the trailing metadata sent by the server. 

137 

138 Returns: 

139 The trailing metadata received. 

140 """ 

141 return self._trailing_metadata 

142 

143 def debug_error_string(self) -> Optional[str]: 

144 """Accesses the debug error string sent by the server. 

145 

146 Returns: 

147 The debug error string received. 

148 """ 

149 return self._debug_error_string 

150 

151 def _repr(self) -> str: 

152 """Assembles the error string for the RPC error.""" 

153 return _NON_OK_CALL_REPRESENTATION.format( 

154 self.__class__.__name__, 

155 self._code, 

156 self._details, 

157 self._debug_error_string, 

158 ) 

159 

160 def __repr__(self) -> str: 

161 return self._repr() 

162 

163 def __str__(self) -> str: 

164 return self._repr() 

165 

166 def __reduce__(self): 

167 return ( 

168 type(self), 

169 ( 

170 self._code, 

171 self._initial_metadata, 

172 self._trailing_metadata, 

173 self._details, 

174 self._debug_error_string, 

175 ), 

176 ) 

177 

178 

179def _create_rpc_error( 

180 initial_metadata: MetadataType, 

181 status: cygrpc.AioRpcStatus, 

182) -> AioRpcError: 

183 return AioRpcError( 

184 _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE[status.code()], 

185 Metadata._create( # pyright: ignore[reportPrivateUsage] 

186 initial_metadata 

187 ), 

188 Metadata.from_tuple(status.trailing_metadata()), 

189 details=status.details(), 

190 debug_error_string=status.debug_error_string(), 

191 ) 

192 

193 

194class Call(Generic[RequestType, ResponseType]): 

195 """Base implementation of client RPC Call object. 

196 

197 Implements logic around final status, metadata and cancellation. 

198 """ 

199 

200 _loop: asyncio.AbstractEventLoop 

201 _code: grpc.StatusCode 

202 _cython_call: cygrpc._AioCall 

203 _metadata: Tuple[MetadatumType, ...] 

204 _request_serializer: Optional[SerializingFunction[RequestType]] 

205 _response_deserializer: Optional[DeserializingFunction[ResponseType]] 

206 

207 def __init__( 

208 self, 

209 cython_call: cygrpc._AioCall, 

210 metadata: Metadata, 

211 request_serializer: Optional[SerializingFunction[RequestType]], 

212 response_deserializer: Optional[DeserializingFunction[ResponseType]], 

213 loop: asyncio.AbstractEventLoop, 

214 ) -> None: 

215 self._loop = loop 

216 self._cython_call = cython_call 

217 self._metadata = tuple(metadata) 

218 self._request_serializer = request_serializer 

219 self._response_deserializer = response_deserializer 

220 

221 def __del__(self) -> None: 

222 # The '_cython_call' object might be destructed before Call object 

223 if hasattr(self, "_cython_call") and not self._cython_call.done(): 

224 self._cancel(_GC_CANCELLATION_DETAILS) 

225 

226 def cancelled(self) -> bool: 

227 return self._cython_call.cancelled() 

228 

229 def _cancel(self, details: str) -> bool: 

230 """Forwards the application cancellation reasoning.""" 

231 if not self._cython_call.done(): 

232 self._cython_call.cancel(details) 

233 return True 

234 return False 

235 

236 def cancel(self) -> bool: 

237 return self._cancel(_LOCAL_CANCELLATION_DETAILS) 

238 

239 def done(self) -> bool: 

240 return self._cython_call.done() 

241 

242 def add_done_callback(self, callback: DoneCallbackType) -> None: 

243 cb = partial(callback, self) 

244 self._cython_call.add_done_callback(cb) 

245 

246 def time_remaining(self) -> Optional[float]: 

247 return self._cython_call.time_remaining() 

248 

249 async def initial_metadata(self) -> Metadata: 

250 raw_metadata_tuple = await self._cython_call.initial_metadata() 

251 return Metadata.from_tuple(raw_metadata_tuple) 

252 

253 async def trailing_metadata(self) -> Metadata: 

254 raw_metadata_tuple = ( 

255 await self._cython_call.status() 

256 ).trailing_metadata() 

257 if not raw_metadata_tuple: 

258 return Metadata() 

259 return Metadata.from_tuple(raw_metadata_tuple) 

260 

261 async def code(self) -> grpc.StatusCode: 

262 cygrpc_code = (await self._cython_call.status()).code() 

263 return _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE[cygrpc_code] 

264 

265 async def details(self) -> str: 

266 return (await self._cython_call.status()).details() 

267 

268 async def debug_error_string(self) -> str: 

269 return (await self._cython_call.status()).debug_error_string() 

270 

271 async def _raise_for_status(self) -> None: 

272 if self._cython_call.is_locally_cancelled(): 

273 raise asyncio.CancelledError() 

274 code = await self.code() 

275 if code != grpc.StatusCode.OK: 

276 raise _create_rpc_error( 

277 await self.initial_metadata(), 

278 await self._cython_call.status(), 

279 ) 

280 

281 def _repr(self) -> str: 

282 return repr(self._cython_call) 

283 

284 def __repr__(self) -> str: 

285 return self._repr() 

286 

287 def __str__(self) -> str: 

288 return self._repr() 

289 

290 

291class _APIStyle(enum.IntEnum): 

292 UNKNOWN = 0 

293 ASYNC_GENERATOR = 1 

294 READER_WRITER = 2 

295 

296 

297class _UnaryResponseMixin(Call[RequestType, ResponseType]): 

298 _call_response: asyncio.Task[Union[ResponseType, EOFType]] 

299 

300 def _init_unary_response_mixin( 

301 self, response_task: asyncio.Task[Union[ResponseType, EOFType]] 

302 ): 

303 self._call_response = response_task 

304 

305 def cancel(self) -> bool: 

306 if super().cancel(): 

307 self._call_response.cancel() 

308 return True 

309 return False 

310 

311 def __await__(self) -> Generator[Any, None, ResponseType]: 

312 """Wait till the ongoing RPC request finishes.""" 

313 try: 

314 response = yield from self._call_response 

315 except asyncio.CancelledError: 

316 # Even if we caught all other CancelledError, there is still 

317 # this corner case. If the application cancels immediately after 

318 # the Call object is created, we will observe this 

319 # `CancelledError`. 

320 if not self.cancelled(): 

321 self.cancel() 

322 raise 

323 

324 # NOTE(lidiz) If we raise RpcError in the task, and users doesn't 

325 # 'await' on it. AsyncIO will log 'Task exception was never retrieved'. 

326 # Instead, if we move the exception raising here, the spam stops. 

327 # Unfortunately, there can only be one 'yield from' in '__await__'. So, 

328 # we need to access the private instance variable. 

329 if response is cygrpc.EOF: 

330 if self._cython_call.is_locally_cancelled(): 

331 raise asyncio.CancelledError() 

332 else: 

333 raise _create_rpc_error( 

334 self._cython_call._initial_metadata, 

335 self._cython_call._status, 

336 ) 

337 else: 

338 return response 

339 

340 

341class _StreamResponseMixin(Call[RequestType, ResponseType]): 

342 _message_aiter: Optional[AsyncIterator[ResponseType]] 

343 _preparation: asyncio.Task[None] 

344 _response_style: _APIStyle 

345 

346 def _init_stream_response_mixin(self, preparation: asyncio.Task[None]): 

347 self._message_aiter = None 

348 self._preparation = preparation 

349 self._response_style = _APIStyle.UNKNOWN 

350 

351 def _update_response_style(self, style: _APIStyle): 

352 if self._response_style is _APIStyle.UNKNOWN: 

353 self._response_style = style 

354 elif self._response_style is not style: 

355 raise cygrpc.UsageError(_API_STYLE_ERROR) 

356 

357 def cancel(self) -> bool: 

358 if super().cancel(): 

359 self._preparation.cancel() 

360 return True 

361 return False 

362 

363 async def _fetch_stream_responses(self) -> AsyncIterator[ResponseType]: 

364 message = await self._read() 

365 while message is not cygrpc.EOF: 

366 yield message 

367 message = await self._read() 

368 

369 # If the read operation failed, Core should explain why. 

370 await self._raise_for_status() 

371 

372 def __aiter__(self) -> AsyncIterator[ResponseType]: 

373 self._update_response_style(_APIStyle.ASYNC_GENERATOR) 

374 if self._message_aiter is None: 

375 self._message_aiter = self._fetch_stream_responses() 

376 return self._message_aiter 

377 

378 async def _read(self) -> Union[ResponseType, EOFType]: 

379 # Wait for the request being sent 

380 await self._preparation 

381 

382 # Reads response message from Core 

383 try: 

384 raw_response = await self._cython_call.receive_serialized_message() 

385 except asyncio.CancelledError: 

386 if not self.cancelled(): 

387 self.cancel() 

388 raise 

389 

390 if raw_response is cygrpc.EOF: 

391 return cygrpc.EOF 

392 return _common.deserialize(raw_response, self._response_deserializer) 

393 

394 async def read(self) -> Union[ResponseType, EOFType]: 

395 if self.done(): 

396 await self._raise_for_status() 

397 return cygrpc.EOF 

398 self._update_response_style(_APIStyle.READER_WRITER) 

399 

400 response_message = await self._read() 

401 

402 if response_message is cygrpc.EOF: 

403 # If the read operation failed, Core should explain why. 

404 await self._raise_for_status() 

405 return response_message 

406 

407 

408class _StreamRequestMixin(Call[RequestType, ResponseType]): 

409 _metadata_sent: asyncio.Event 

410 _done_writing_flag: bool 

411 _async_request_poller: Optional[asyncio.Task[None]] 

412 _request_style: _APIStyle 

413 

414 def _init_stream_request_mixin( 

415 self, request_iterator: Optional[RequestIterableType[RequestType]] 

416 ): 

417 self._metadata_sent = asyncio.Event() 

418 self._done_writing_flag = False 

419 

420 # If user passes in an async iterator, create a consumer Task. 

421 if request_iterator is not None: 

422 self._async_request_poller = self._loop.create_task( 

423 self._consume_request_iterator(request_iterator) 

424 ) 

425 self._request_style = _APIStyle.ASYNC_GENERATOR 

426 else: 

427 self._async_request_poller = None 

428 self._request_style = _APIStyle.READER_WRITER 

429 

430 def _raise_for_different_style(self, style: _APIStyle): 

431 if self._request_style is not style: 

432 raise cygrpc.UsageError(_API_STYLE_ERROR) 

433 

434 def cancel(self) -> bool: 

435 if super().cancel(): 

436 if self._async_request_poller is not None: 

437 self._async_request_poller.cancel() 

438 return True 

439 return False 

440 

441 def _metadata_sent_observer(self): 

442 self._metadata_sent.set() 

443 

444 async def _consume_request_iterator( 

445 self, request_iterator: RequestIterableType[RequestType] 

446 ) -> None: 

447 try: 

448 if isinstance(request_iterator, AsyncIterable): 

449 async for request in request_iterator: 

450 try: 

451 await self._write(request) 

452 except AioRpcError as rpc_error: 

453 _LOGGER.debug( 

454 ( 

455 "Exception while consuming the" 

456 " request_iterator: %s" 

457 ), 

458 rpc_error, 

459 ) 

460 return 

461 else: 

462 for request in request_iterator: 

463 try: 

464 await self._write(request) 

465 except AioRpcError as rpc_error: 

466 _LOGGER.debug( 

467 ( 

468 "Exception while consuming the" 

469 " request_iterator: %s" 

470 ), 

471 rpc_error, 

472 ) 

473 return 

474 

475 await self._done_writing() 

476 except: # pylint: disable=bare-except # noqa: E722 

477 # Client iterators can raise exceptions, which we should handle by 

478 # cancelling the RPC and logging the client's error. No exceptions 

479 # should escape this function. 

480 _LOGGER.debug( 

481 "Client request_iterator raised exception:\n%s", 

482 traceback.format_exc(), 

483 ) 

484 self.cancel() 

485 

486 async def _write(self, request: RequestType) -> None: 

487 if self.done(): 

488 raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS) 

489 if self._done_writing_flag: 

490 raise asyncio.InvalidStateError(_RPC_HALF_CLOSED_DETAILS) 

491 if not self._metadata_sent.is_set(): 

492 await self._metadata_sent.wait() 

493 if self.done(): 

494 await self._raise_for_status() 

495 

496 serialized_request = _common.serialize( 

497 request, self._request_serializer 

498 ) 

499 try: 

500 await self._cython_call.send_serialized_message(serialized_request) 

501 except cygrpc.InternalError as err: 

502 self._cython_call.set_internal_error(str(err)) 

503 await self._raise_for_status() 

504 except asyncio.CancelledError: 

505 if not self.cancelled(): 

506 self.cancel() 

507 raise 

508 

509 async def _done_writing(self) -> None: 

510 if self.done(): 

511 # If the RPC is finished, do nothing. 

512 return 

513 if not self._done_writing_flag: 

514 # If the done writing is not sent before, try to send it. 

515 self._done_writing_flag = True 

516 try: 

517 await self._cython_call.send_receive_close() 

518 except asyncio.CancelledError: 

519 if not self.cancelled(): 

520 self.cancel() 

521 raise 

522 

523 async def write(self, request: RequestType) -> None: 

524 self._raise_for_different_style(_APIStyle.READER_WRITER) 

525 await self._write(request) 

526 

527 async def done_writing(self) -> None: 

528 """Signal peer that client is done writing. 

529 

530 This method is idempotent. 

531 """ 

532 self._raise_for_different_style(_APIStyle.READER_WRITER) 

533 await self._done_writing() 

534 

535 async def wait_for_connection(self) -> None: 

536 await self._metadata_sent.wait() 

537 if self.done(): 

538 await self._raise_for_status() 

539 

540 

541class UnaryUnaryCall( 

542 _UnaryResponseMixin[RequestType, ResponseType], 

543 _base_call.UnaryUnaryCall[RequestType, ResponseType], 

544): 

545 """Object for managing unary-unary RPC calls. 

546 

547 Returned when an instance of `UnaryUnaryMultiCallable` object is called. 

548 """ 

549 

550 _request: RequestType 

551 _invocation_task: asyncio.Task[Union[ResponseType, EOFType]] 

552 

553 # pylint: disable=too-many-arguments 

554 def __init__( 

555 self, 

556 request: RequestType, 

557 deadline: Optional[float], 

558 metadata: Metadata, 

559 credentials: Optional[grpc.CallCredentials], 

560 wait_for_ready: Optional[bool], 

561 channel: cygrpc.AioChannel, 

562 method: bytes, 

563 request_serializer: Optional[SerializingFunction[RequestType]], 

564 response_deserializer: Optional[DeserializingFunction[ResponseType]], 

565 loop: asyncio.AbstractEventLoop, 

566 registered_call_handle: int = 0, 

567 ) -> None: 

568 super().__init__( 

569 channel.call( 

570 method, 

571 deadline, 

572 credentials, 

573 wait_for_ready, 

574 registered_call_handle, 

575 ), 

576 metadata, 

577 request_serializer, 

578 response_deserializer, 

579 loop, 

580 ) 

581 self._request = request 

582 self._context = cygrpc.build_census_context() 

583 self._invocation_task = loop.create_task(self._invoke()) 

584 self._init_unary_response_mixin(self._invocation_task) 

585 

586 async def _invoke(self) -> Union[ResponseType, EOFType]: 

587 serialized_request = _common.serialize( 

588 self._request, self._request_serializer 

589 ) 

590 

591 # NOTE(lidiz) asyncio.CancelledError is not a good transport for status, 

592 # because the asyncio.Task class do not cache the exception object. 

593 # https://github.com/python/cpython/blob/edad4d89e357c92f70c0324b937845d652b20afd/Lib/asyncio/tasks.py#L785 

594 try: 

595 serialized_response = await self._cython_call.unary_unary( 

596 serialized_request, self._metadata, self._context 

597 ) 

598 except asyncio.CancelledError: 

599 if not self.cancelled(): 

600 self.cancel() 

601 return cygrpc.EOF 

602 

603 if self._cython_call.is_ok(): 

604 return _common.deserialize( 

605 serialized_response, self._response_deserializer 

606 ) 

607 return cygrpc.EOF 

608 

609 async def wait_for_connection(self) -> None: 

610 await self._invocation_task 

611 if self.done(): 

612 await self._raise_for_status() 

613 

614 

615class UnaryStreamCall( 

616 _StreamResponseMixin[RequestType, ResponseType], 

617 _base_call.UnaryStreamCall[RequestType, ResponseType], 

618): 

619 """Object for managing unary-stream RPC calls. 

620 

621 Returned when an instance of `UnaryStreamMultiCallable` object is called. 

622 """ 

623 

624 _request: RequestType 

625 _send_unary_request_task: asyncio.Task[None] 

626 

627 # pylint: disable=too-many-arguments 

628 def __init__( 

629 self, 

630 request: RequestType, 

631 deadline: Optional[float], 

632 metadata: Metadata, 

633 credentials: Optional[grpc.CallCredentials], 

634 wait_for_ready: Optional[bool], 

635 channel: cygrpc.AioChannel, 

636 method: bytes, 

637 request_serializer: Optional[SerializingFunction[RequestType]], 

638 response_deserializer: Optional[DeserializingFunction[ResponseType]], 

639 loop: asyncio.AbstractEventLoop, 

640 registered_call_handle: int = 0, 

641 ) -> None: 

642 super().__init__( 

643 channel.call( 

644 method, 

645 deadline, 

646 credentials, 

647 wait_for_ready, 

648 registered_call_handle, 

649 ), 

650 metadata, 

651 request_serializer, 

652 response_deserializer, 

653 loop, 

654 ) 

655 self._request = request 

656 self._context = cygrpc.build_census_context() 

657 self._send_unary_request_task = loop.create_task( 

658 self._send_unary_request() 

659 ) 

660 self._init_stream_response_mixin(self._send_unary_request_task) 

661 

662 async def _send_unary_request(self) -> None: 

663 serialized_request = _common.serialize( 

664 self._request, self._request_serializer 

665 ) 

666 try: 

667 await self._cython_call.initiate_unary_stream( 

668 serialized_request, self._metadata, self._context 

669 ) 

670 except asyncio.CancelledError: 

671 if not self.cancelled(): 

672 self.cancel() 

673 raise 

674 

675 async def wait_for_connection(self) -> None: 

676 await self._send_unary_request_task 

677 if self.done(): 

678 await self._raise_for_status() 

679 

680 

681# pylint: disable=too-many-ancestors 

682class StreamUnaryCall( 

683 _StreamRequestMixin[RequestType, ResponseType], 

684 _UnaryResponseMixin[RequestType, ResponseType], 

685 _base_call.StreamUnaryCall[RequestType, ResponseType], 

686): 

687 """Object for managing stream-unary RPC calls. 

688 

689 Returned when an instance of `StreamUnaryMultiCallable` object is called. 

690 """ 

691 

692 # pylint: disable=too-many-arguments 

693 def __init__( 

694 self, 

695 request_iterator: Optional[RequestIterableType[RequestType]], 

696 deadline: Optional[float], 

697 metadata: Metadata, 

698 credentials: Optional[grpc.CallCredentials], 

699 wait_for_ready: Optional[bool], 

700 channel: cygrpc.AioChannel, 

701 method: bytes, 

702 request_serializer: Optional[SerializingFunction[RequestType]], 

703 response_deserializer: Optional[DeserializingFunction[ResponseType]], 

704 loop: asyncio.AbstractEventLoop, 

705 registered_call_handle: int = 0, 

706 ) -> None: 

707 super().__init__( 

708 channel.call( 

709 method, 

710 deadline, 

711 credentials, 

712 wait_for_ready, 

713 registered_call_handle, 

714 ), 

715 metadata, 

716 request_serializer, 

717 response_deserializer, 

718 loop, 

719 ) 

720 

721 self._context = cygrpc.build_census_context() 

722 self._init_stream_request_mixin(request_iterator) 

723 self._init_unary_response_mixin(loop.create_task(self._conduct_rpc())) 

724 

725 async def _conduct_rpc(self) -> Union[ResponseType, EOFType]: 

726 try: 

727 serialized_response = await self._cython_call.stream_unary( 

728 self._metadata, self._metadata_sent_observer, self._context 

729 ) 

730 except asyncio.CancelledError: 

731 if not self.cancelled(): 

732 self.cancel() 

733 raise 

734 

735 if self._cython_call.is_ok(): 

736 return _common.deserialize( 

737 serialized_response, self._response_deserializer 

738 ) 

739 return cygrpc.EOF 

740 

741 

742class StreamStreamCall( 

743 _StreamRequestMixin[RequestType, ResponseType], 

744 _StreamResponseMixin[RequestType, ResponseType], 

745 _base_call.StreamStreamCall[RequestType, ResponseType], 

746): 

747 """Object for managing stream-stream RPC calls. 

748 

749 Returned when an instance of `StreamStreamMultiCallable` object is called. 

750 """ 

751 

752 _initializer: asyncio.Task[None] 

753 

754 # pylint: disable=too-many-arguments 

755 def __init__( 

756 self, 

757 request_iterator: Optional[RequestIterableType[RequestType]], 

758 deadline: Optional[float], 

759 metadata: Metadata, 

760 credentials: Optional[grpc.CallCredentials], 

761 wait_for_ready: Optional[bool], 

762 channel: cygrpc.AioChannel, 

763 method: bytes, 

764 request_serializer: Optional[SerializingFunction[RequestType]], 

765 response_deserializer: Optional[DeserializingFunction[ResponseType]], 

766 loop: asyncio.AbstractEventLoop, 

767 registered_call_handle: int = 0, 

768 ) -> None: 

769 super().__init__( 

770 channel.call( 

771 method, 

772 deadline, 

773 credentials, 

774 wait_for_ready, 

775 registered_call_handle, 

776 ), 

777 metadata, 

778 request_serializer, 

779 response_deserializer, 

780 loop, 

781 ) 

782 self._context = cygrpc.build_census_context() 

783 self._initializer = self._loop.create_task(self._prepare_rpc()) 

784 self._init_stream_request_mixin(request_iterator) 

785 self._init_stream_response_mixin(self._initializer) 

786 

787 async def _prepare_rpc(self) -> None: 

788 """Prepares the RPC for receiving/sending messages. 

789 

790 All other operations around the stream should only happen after the 

791 completion of this method. 

792 """ 

793 try: 

794 await self._cython_call.initiate_stream_stream( 

795 self._metadata, self._metadata_sent_observer, self._context 

796 ) 

797 except asyncio.CancelledError: 

798 if not self.cancelled(): 

799 self.cancel() 

800 # No need to raise RpcError here, because no one will `await` this task.