Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/pubsub_v1/services/schema_service/transports/rest_base.py: 52%

1# -*- coding: utf-8 -*- 

2# Copyright 2025 Google LLC 

3# 

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

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

6# You may obtain a copy of the License at 

7# 

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

9# 

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

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

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

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

14# limitations under the License. 

15# 

16import json # type: ignore 

17from google.api_core import path_template 

18from google.api_core import gapic_v1 

19 

20from google.protobuf import json_format 

21from google.iam.v1 import iam_policy_pb2 # type: ignore 

22from google.iam.v1 import policy_pb2 # type: ignore 

23from .base import SchemaServiceTransport, DEFAULT_CLIENT_INFO 

24 

25import re 

26from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union 

27 

28 

29from google.protobuf import empty_pb2 # type: ignore 

30from google.pubsub_v1.types import schema 

31from google.pubsub_v1.types import schema as gp_schema 

32 

33 

34class _BaseSchemaServiceRestTransport(SchemaServiceTransport): 

35 """Base REST backend transport for SchemaService. 

36 

37 Note: This class is not meant to be used directly. Use its sync and 

38 async sub-classes instead. 

39 

40 This class defines the same methods as the primary client, so the 

41 primary client can load the underlying transport implementation 

42 and call it. 

43 

44 It sends JSON representations of protocol buffers over HTTP/1.1 

45 """ 

46 

47 def __init__( 

48 self, 

49 *, 

50 host: str = "pubsub.googleapis.com", 

51 credentials: Optional[Any] = None, 

52 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

53 always_use_jwt_access: Optional[bool] = False, 

54 url_scheme: str = "https", 

55 api_audience: Optional[str] = None, 

56 ) -> None: 

57 """Instantiate the transport. 

58 Args: 

59 host (Optional[str]): 

60 The hostname to connect to (default: 'pubsub.googleapis.com'). 

61 credentials (Optional[Any]): The 

62 authorization credentials to attach to requests. These 

63 credentials identify the application to the service; if none 

64 are specified, the client will attempt to ascertain the 

65 credentials from the environment. 

66 client_info (google.api_core.gapic_v1.client_info.ClientInfo): 

67 The client info used to send a user-agent string along with 

68 API requests. If ``None``, then default info will be used. 

69 Generally, you only need to set this if you are developing 

70 your own client library. 

71 always_use_jwt_access (Optional[bool]): Whether self signed JWT should 

72 be used for service account credentials. 

73 url_scheme: the protocol scheme for the API endpoint. Normally 

74 "https", but for testing or local servers, 

75 "http" can be specified. 

76 """ 

77 # Run the base constructor 

78 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host) 

79 if maybe_url_match is None: 

80 raise ValueError( 

81 f"Unexpected hostname structure: {host}" 

82 ) # pragma: NO COVER 

83 

84 url_match_items = maybe_url_match.groupdict() 

85 

86 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host 

87 

88 super().__init__( 

89 host=host, 

90 credentials=credentials, 

91 client_info=client_info, 

92 always_use_jwt_access=always_use_jwt_access, 

93 api_audience=api_audience, 

94 ) 

95 

96 class _BaseCommitSchema: 

97 def __hash__(self): # pragma: NO COVER 

98 return NotImplementedError("__hash__ must be implemented.") 

99 

100 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

101 

102 @classmethod 

103 def _get_unset_required_fields(cls, message_dict): 

104 return { 

105 k: v 

106 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

107 if k not in message_dict 

108 } 

109 

110 @staticmethod 

111 def _get_http_options(): 

112 http_options: List[Dict[str, str]] = [ 

113 { 

114 "method": "post", 

115 "uri": "/v1/{name=projects/*/schemas/*}:commit", 

116 "body": "*", 

117 }, 

118 ] 

119 return http_options 

120 

121 @staticmethod 

122 def _get_transcoded_request(http_options, request): 

123 pb_request = gp_schema.CommitSchemaRequest.pb(request) 

124 transcoded_request = path_template.transcode(http_options, pb_request) 

125 return transcoded_request 

126 

127 @staticmethod 

128 def _get_request_body_json(transcoded_request): 

129 # Jsonify the request body 

130 

131 body = json_format.MessageToJson( 

132 transcoded_request["body"], use_integers_for_enums=True 

133 ) 

134 return body 

135 

136 @staticmethod 

137 def _get_query_params_json(transcoded_request): 

138 query_params = json.loads( 

139 json_format.MessageToJson( 

140 transcoded_request["query_params"], 

141 use_integers_for_enums=True, 

142 ) 

143 ) 

144 query_params.update( 

145 _BaseSchemaServiceRestTransport._BaseCommitSchema._get_unset_required_fields( 

146 query_params 

147 ) 

148 ) 

149 

150 query_params["$alt"] = "json;enum-encoding=int" 

151 return query_params 

152 

153 class _BaseCreateSchema: 

154 def __hash__(self): # pragma: NO COVER 

155 return NotImplementedError("__hash__ must be implemented.") 

156 

157 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

158 

159 @classmethod 

160 def _get_unset_required_fields(cls, message_dict): 

161 return { 

162 k: v 

163 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

164 if k not in message_dict 

165 } 

166 

167 @staticmethod 

168 def _get_http_options(): 

169 http_options: List[Dict[str, str]] = [ 

170 { 

171 "method": "post", 

172 "uri": "/v1/{parent=projects/*}/schemas", 

173 "body": "schema", 

174 }, 

175 ] 

176 return http_options 

177 

178 @staticmethod 

179 def _get_transcoded_request(http_options, request): 

180 pb_request = gp_schema.CreateSchemaRequest.pb(request) 

181 transcoded_request = path_template.transcode(http_options, pb_request) 

182 return transcoded_request 

183 

184 @staticmethod 

185 def _get_request_body_json(transcoded_request): 

186 # Jsonify the request body 

187 

188 body = json_format.MessageToJson( 

189 transcoded_request["body"], use_integers_for_enums=True 

190 ) 

191 return body 

192 

193 @staticmethod 

194 def _get_query_params_json(transcoded_request): 

195 query_params = json.loads( 

196 json_format.MessageToJson( 

197 transcoded_request["query_params"], 

198 use_integers_for_enums=True, 

199 ) 

200 ) 

201 query_params.update( 

202 _BaseSchemaServiceRestTransport._BaseCreateSchema._get_unset_required_fields( 

203 query_params 

204 ) 

205 ) 

206 

207 query_params["$alt"] = "json;enum-encoding=int" 

208 return query_params 

209 

210 class _BaseDeleteSchema: 

211 def __hash__(self): # pragma: NO COVER 

212 return NotImplementedError("__hash__ must be implemented.") 

213 

214 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

215 

216 @classmethod 

217 def _get_unset_required_fields(cls, message_dict): 

218 return { 

219 k: v 

220 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

221 if k not in message_dict 

222 } 

223 

224 @staticmethod 

225 def _get_http_options(): 

226 http_options: List[Dict[str, str]] = [ 

227 { 

228 "method": "delete", 

229 "uri": "/v1/{name=projects/*/schemas/*}", 

230 }, 

231 ] 

232 return http_options 

233 

234 @staticmethod 

235 def _get_transcoded_request(http_options, request): 

236 pb_request = schema.DeleteSchemaRequest.pb(request) 

237 transcoded_request = path_template.transcode(http_options, pb_request) 

238 return transcoded_request 

239 

240 @staticmethod 

241 def _get_query_params_json(transcoded_request): 

242 query_params = json.loads( 

243 json_format.MessageToJson( 

244 transcoded_request["query_params"], 

245 use_integers_for_enums=True, 

246 ) 

247 ) 

248 query_params.update( 

249 _BaseSchemaServiceRestTransport._BaseDeleteSchema._get_unset_required_fields( 

250 query_params 

251 ) 

252 ) 

253 

254 query_params["$alt"] = "json;enum-encoding=int" 

255 return query_params 

256 

257 class _BaseDeleteSchemaRevision: 

258 def __hash__(self): # pragma: NO COVER 

259 return NotImplementedError("__hash__ must be implemented.") 

260 

261 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

262 

263 @classmethod 

264 def _get_unset_required_fields(cls, message_dict): 

265 return { 

266 k: v 

267 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

268 if k not in message_dict 

269 } 

270 

271 @staticmethod 

272 def _get_http_options(): 

273 http_options: List[Dict[str, str]] = [ 

274 { 

275 "method": "delete", 

276 "uri": "/v1/{name=projects/*/schemas/*}:deleteRevision", 

277 }, 

278 ] 

279 return http_options 

280 

281 @staticmethod 

282 def _get_transcoded_request(http_options, request): 

283 pb_request = schema.DeleteSchemaRevisionRequest.pb(request) 

284 transcoded_request = path_template.transcode(http_options, pb_request) 

285 return transcoded_request 

286 

287 @staticmethod 

288 def _get_query_params_json(transcoded_request): 

289 query_params = json.loads( 

290 json_format.MessageToJson( 

291 transcoded_request["query_params"], 

292 use_integers_for_enums=True, 

293 ) 

294 ) 

295 query_params.update( 

296 _BaseSchemaServiceRestTransport._BaseDeleteSchemaRevision._get_unset_required_fields( 

297 query_params 

298 ) 

299 ) 

300 

301 query_params["$alt"] = "json;enum-encoding=int" 

302 return query_params 

303 

304 class _BaseGetSchema: 

305 def __hash__(self): # pragma: NO COVER 

306 return NotImplementedError("__hash__ must be implemented.") 

307 

308 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

309 

310 @classmethod 

311 def _get_unset_required_fields(cls, message_dict): 

312 return { 

313 k: v 

314 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

315 if k not in message_dict 

316 } 

317 

318 @staticmethod 

319 def _get_http_options(): 

320 http_options: List[Dict[str, str]] = [ 

321 { 

322 "method": "get", 

323 "uri": "/v1/{name=projects/*/schemas/*}", 

324 }, 

325 ] 

326 return http_options 

327 

328 @staticmethod 

329 def _get_transcoded_request(http_options, request): 

330 pb_request = schema.GetSchemaRequest.pb(request) 

331 transcoded_request = path_template.transcode(http_options, pb_request) 

332 return transcoded_request 

333 

334 @staticmethod 

335 def _get_query_params_json(transcoded_request): 

336 query_params = json.loads( 

337 json_format.MessageToJson( 

338 transcoded_request["query_params"], 

339 use_integers_for_enums=True, 

340 ) 

341 ) 

342 query_params.update( 

343 _BaseSchemaServiceRestTransport._BaseGetSchema._get_unset_required_fields( 

344 query_params 

345 ) 

346 ) 

347 

348 query_params["$alt"] = "json;enum-encoding=int" 

349 return query_params 

350 

351 class _BaseListSchemaRevisions: 

352 def __hash__(self): # pragma: NO COVER 

353 return NotImplementedError("__hash__ must be implemented.") 

354 

355 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

356 

357 @classmethod 

358 def _get_unset_required_fields(cls, message_dict): 

359 return { 

360 k: v 

361 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

362 if k not in message_dict 

363 } 

364 

365 @staticmethod 

366 def _get_http_options(): 

367 http_options: List[Dict[str, str]] = [ 

368 { 

369 "method": "get", 

370 "uri": "/v1/{name=projects/*/schemas/*}:listRevisions", 

371 }, 

372 ] 

373 return http_options 

374 

375 @staticmethod 

376 def _get_transcoded_request(http_options, request): 

377 pb_request = schema.ListSchemaRevisionsRequest.pb(request) 

378 transcoded_request = path_template.transcode(http_options, pb_request) 

379 return transcoded_request 

380 

381 @staticmethod 

382 def _get_query_params_json(transcoded_request): 

383 query_params = json.loads( 

384 json_format.MessageToJson( 

385 transcoded_request["query_params"], 

386 use_integers_for_enums=True, 

387 ) 

388 ) 

389 query_params.update( 

390 _BaseSchemaServiceRestTransport._BaseListSchemaRevisions._get_unset_required_fields( 

391 query_params 

392 ) 

393 ) 

394 

395 query_params["$alt"] = "json;enum-encoding=int" 

396 return query_params 

397 

398 class _BaseListSchemas: 

399 def __hash__(self): # pragma: NO COVER 

400 return NotImplementedError("__hash__ must be implemented.") 

401 

402 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

403 

404 @classmethod 

405 def _get_unset_required_fields(cls, message_dict): 

406 return { 

407 k: v 

408 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

409 if k not in message_dict 

410 } 

411 

412 @staticmethod 

413 def _get_http_options(): 

414 http_options: List[Dict[str, str]] = [ 

415 { 

416 "method": "get", 

417 "uri": "/v1/{parent=projects/*}/schemas", 

418 }, 

419 ] 

420 return http_options 

421 

422 @staticmethod 

423 def _get_transcoded_request(http_options, request): 

424 pb_request = schema.ListSchemasRequest.pb(request) 

425 transcoded_request = path_template.transcode(http_options, pb_request) 

426 return transcoded_request 

427 

428 @staticmethod 

429 def _get_query_params_json(transcoded_request): 

430 query_params = json.loads( 

431 json_format.MessageToJson( 

432 transcoded_request["query_params"], 

433 use_integers_for_enums=True, 

434 ) 

435 ) 

436 query_params.update( 

437 _BaseSchemaServiceRestTransport._BaseListSchemas._get_unset_required_fields( 

438 query_params 

439 ) 

440 ) 

441 

442 query_params["$alt"] = "json;enum-encoding=int" 

443 return query_params 

444 

445 class _BaseRollbackSchema: 

446 def __hash__(self): # pragma: NO COVER 

447 return NotImplementedError("__hash__ must be implemented.") 

448 

449 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

450 

451 @classmethod 

452 def _get_unset_required_fields(cls, message_dict): 

453 return { 

454 k: v 

455 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

456 if k not in message_dict 

457 } 

458 

459 @staticmethod 

460 def _get_http_options(): 

461 http_options: List[Dict[str, str]] = [ 

462 { 

463 "method": "post", 

464 "uri": "/v1/{name=projects/*/schemas/*}:rollback", 

465 "body": "*", 

466 }, 

467 ] 

468 return http_options 

469 

470 @staticmethod 

471 def _get_transcoded_request(http_options, request): 

472 pb_request = schema.RollbackSchemaRequest.pb(request) 

473 transcoded_request = path_template.transcode(http_options, pb_request) 

474 return transcoded_request 

475 

476 @staticmethod 

477 def _get_request_body_json(transcoded_request): 

478 # Jsonify the request body 

479 

480 body = json_format.MessageToJson( 

481 transcoded_request["body"], use_integers_for_enums=True 

482 ) 

483 return body 

484 

485 @staticmethod 

486 def _get_query_params_json(transcoded_request): 

487 query_params = json.loads( 

488 json_format.MessageToJson( 

489 transcoded_request["query_params"], 

490 use_integers_for_enums=True, 

491 ) 

492 ) 

493 query_params.update( 

494 _BaseSchemaServiceRestTransport._BaseRollbackSchema._get_unset_required_fields( 

495 query_params 

496 ) 

497 ) 

498 

499 query_params["$alt"] = "json;enum-encoding=int" 

500 return query_params 

501 

502 class _BaseValidateMessage: 

503 def __hash__(self): # pragma: NO COVER 

504 return NotImplementedError("__hash__ must be implemented.") 

505 

506 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

507 

508 @classmethod 

509 def _get_unset_required_fields(cls, message_dict): 

510 return { 

511 k: v 

512 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

513 if k not in message_dict 

514 } 

515 

516 @staticmethod 

517 def _get_http_options(): 

518 http_options: List[Dict[str, str]] = [ 

519 { 

520 "method": "post", 

521 "uri": "/v1/{parent=projects/*}/schemas:validateMessage", 

522 "body": "*", 

523 }, 

524 ] 

525 return http_options 

526 

527 @staticmethod 

528 def _get_transcoded_request(http_options, request): 

529 pb_request = schema.ValidateMessageRequest.pb(request) 

530 transcoded_request = path_template.transcode(http_options, pb_request) 

531 return transcoded_request 

532 

533 @staticmethod 

534 def _get_request_body_json(transcoded_request): 

535 # Jsonify the request body 

536 

537 body = json_format.MessageToJson( 

538 transcoded_request["body"], use_integers_for_enums=True 

539 ) 

540 return body 

541 

542 @staticmethod 

543 def _get_query_params_json(transcoded_request): 

544 query_params = json.loads( 

545 json_format.MessageToJson( 

546 transcoded_request["query_params"], 

547 use_integers_for_enums=True, 

548 ) 

549 ) 

550 query_params.update( 

551 _BaseSchemaServiceRestTransport._BaseValidateMessage._get_unset_required_fields( 

552 query_params 

553 ) 

554 ) 

555 

556 query_params["$alt"] = "json;enum-encoding=int" 

557 return query_params 

558 

559 class _BaseValidateSchema: 

560 def __hash__(self): # pragma: NO COVER 

561 return NotImplementedError("__hash__ must be implemented.") 

562 

563 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

564 

565 @classmethod 

566 def _get_unset_required_fields(cls, message_dict): 

567 return { 

568 k: v 

569 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

570 if k not in message_dict 

571 } 

572 

573 @staticmethod 

574 def _get_http_options(): 

575 http_options: List[Dict[str, str]] = [ 

576 { 

577 "method": "post", 

578 "uri": "/v1/{parent=projects/*}/schemas:validate", 

579 "body": "*", 

580 }, 

581 ] 

582 return http_options 

583 

584 @staticmethod 

585 def _get_transcoded_request(http_options, request): 

586 pb_request = gp_schema.ValidateSchemaRequest.pb(request) 

587 transcoded_request = path_template.transcode(http_options, pb_request) 

588 return transcoded_request 

589 

590 @staticmethod 

591 def _get_request_body_json(transcoded_request): 

592 # Jsonify the request body 

593 

594 body = json_format.MessageToJson( 

595 transcoded_request["body"], use_integers_for_enums=True 

596 ) 

597 return body 

598 

599 @staticmethod 

600 def _get_query_params_json(transcoded_request): 

601 query_params = json.loads( 

602 json_format.MessageToJson( 

603 transcoded_request["query_params"], 

604 use_integers_for_enums=True, 

605 ) 

606 ) 

607 query_params.update( 

608 _BaseSchemaServiceRestTransport._BaseValidateSchema._get_unset_required_fields( 

609 query_params 

610 ) 

611 ) 

612 

613 query_params["$alt"] = "json;enum-encoding=int" 

614 return query_params 

615 

616 class _BaseGetIamPolicy: 

617 def __hash__(self): # pragma: NO COVER 

618 return NotImplementedError("__hash__ must be implemented.") 

619 

620 @staticmethod 

621 def _get_http_options(): 

622 http_options: List[Dict[str, str]] = [ 

623 { 

624 "method": "get", 

625 "uri": "/v1/{resource=projects/*/topics/*}:getIamPolicy", 

626 }, 

627 { 

628 "method": "get", 

629 "uri": "/v1/{resource=projects/*/subscriptions/*}:getIamPolicy", 

630 }, 

631 { 

632 "method": "get", 

633 "uri": "/v1/{resource=projects/*/snapshots/*}:getIamPolicy", 

634 }, 

635 { 

636 "method": "get", 

637 "uri": "/v1/{resource=projects/*/schemas/*}:getIamPolicy", 

638 }, 

639 ] 

640 return http_options 

641 

642 @staticmethod 

643 def _get_transcoded_request(http_options, request): 

644 request_kwargs = json_format.MessageToDict(request) 

645 transcoded_request = path_template.transcode(http_options, **request_kwargs) 

646 return transcoded_request 

647 

648 @staticmethod 

649 def _get_query_params_json(transcoded_request): 

650 query_params = json.loads(json.dumps(transcoded_request["query_params"])) 

651 return query_params 

652 

653 class _BaseSetIamPolicy: 

654 def __hash__(self): # pragma: NO COVER 

655 return NotImplementedError("__hash__ must be implemented.") 

656 

657 @staticmethod 

658 def _get_http_options(): 

659 http_options: List[Dict[str, str]] = [ 

660 { 

661 "method": "post", 

662 "uri": "/v1/{resource=projects/*/topics/*}:setIamPolicy", 

663 "body": "*", 

664 }, 

665 { 

666 "method": "post", 

667 "uri": "/v1/{resource=projects/*/subscriptions/*}:setIamPolicy", 

668 "body": "*", 

669 }, 

670 { 

671 "method": "post", 

672 "uri": "/v1/{resource=projects/*/snapshots/*}:setIamPolicy", 

673 "body": "*", 

674 }, 

675 { 

676 "method": "post", 

677 "uri": "/v1/{resource=projects/*/schemas/*}:setIamPolicy", 

678 "body": "*", 

679 }, 

680 ] 

681 return http_options 

682 

683 @staticmethod 

684 def _get_transcoded_request(http_options, request): 

685 request_kwargs = json_format.MessageToDict(request) 

686 transcoded_request = path_template.transcode(http_options, **request_kwargs) 

687 return transcoded_request 

688 

689 @staticmethod 

690 def _get_request_body_json(transcoded_request): 

691 body = json.dumps(transcoded_request["body"]) 

692 return body 

693 

694 @staticmethod 

695 def _get_query_params_json(transcoded_request): 

696 query_params = json.loads(json.dumps(transcoded_request["query_params"])) 

697 return query_params 

698 

699 class _BaseTestIamPermissions: 

700 def __hash__(self): # pragma: NO COVER 

701 return NotImplementedError("__hash__ must be implemented.") 

702 

703 @staticmethod 

704 def _get_http_options(): 

705 http_options: List[Dict[str, str]] = [ 

706 { 

707 "method": "post", 

708 "uri": "/v1/{resource=projects/*/subscriptions/*}:testIamPermissions", 

709 "body": "*", 

710 }, 

711 { 

712 "method": "post", 

713 "uri": "/v1/{resource=projects/*/topics/*}:testIamPermissions", 

714 "body": "*", 

715 }, 

716 { 

717 "method": "post", 

718 "uri": "/v1/{resource=projects/*/snapshots/*}:testIamPermissions", 

719 "body": "*", 

720 }, 

721 { 

722 "method": "post", 

723 "uri": "/v1/{resource=projects/*/schemas/*}:testIamPermissions", 

724 "body": "*", 

725 }, 

726 ] 

727 return http_options 

728 

729 @staticmethod 

730 def _get_transcoded_request(http_options, request): 

731 request_kwargs = json_format.MessageToDict(request) 

732 transcoded_request = path_template.transcode(http_options, **request_kwargs) 

733 return transcoded_request 

734 

735 @staticmethod 

736 def _get_request_body_json(transcoded_request): 

737 body = json.dumps(transcoded_request["body"]) 

738 return body 

739 

740 @staticmethod 

741 def _get_query_params_json(transcoded_request): 

742 query_params = json.loads(json.dumps(transcoded_request["query_params"])) 

743 return query_params 

744 

745 

746__all__ = ("_BaseSchemaServiceRestTransport",)