Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/logging_v2/services/logging_service_v2/transports/grpc.py: 45%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

110 statements  

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 

17import logging as std_logging 

18import pickle 

19import warnings 

20from typing import Callable, Dict, Optional, Sequence, Tuple, Union 

21 

22from google.api_core import grpc_helpers 

23from google.api_core import gapic_v1 

24import google.auth # type: ignore 

25from google.auth import credentials as ga_credentials # type: ignore 

26from google.auth.transport.grpc import SslCredentials # type: ignore 

27from google.protobuf.json_format import MessageToJson 

28import google.protobuf.message 

29 

30import grpc # type: ignore 

31import proto # type: ignore 

32 

33from google.cloud.logging_v2.types import logging 

34from google.longrunning import operations_pb2 # type: ignore 

35from google.protobuf import empty_pb2 # type: ignore 

36from .base import LoggingServiceV2Transport, DEFAULT_CLIENT_INFO 

37 

38try: 

39 from google.api_core import client_logging # type: ignore 

40 

41 CLIENT_LOGGING_SUPPORTED = True # pragma: NO COVER 

42except ImportError: # pragma: NO COVER 

43 CLIENT_LOGGING_SUPPORTED = False 

44 

45_LOGGER = std_logging.getLogger(__name__) 

46 

47 

48class _LoggingClientInterceptor(grpc.UnaryUnaryClientInterceptor): # pragma: NO COVER 

49 def intercept_unary_unary(self, continuation, client_call_details, request): 

50 logging_enabled = CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor( 

51 std_logging.DEBUG 

52 ) 

53 if logging_enabled: # pragma: NO COVER 

54 request_metadata = client_call_details.metadata 

55 if isinstance(request, proto.Message): 

56 request_payload = type(request).to_json(request) 

57 elif isinstance(request, google.protobuf.message.Message): 

58 request_payload = MessageToJson(request) 

59 else: 

60 request_payload = f"{type(request).__name__}: {pickle.dumps(request)}" 

61 

62 request_metadata = { 

63 key: value.decode("utf-8") if isinstance(value, bytes) else value 

64 for key, value in request_metadata 

65 } 

66 grpc_request = { 

67 "payload": request_payload, 

68 "requestMethod": "grpc", 

69 "metadata": dict(request_metadata), 

70 } 

71 _LOGGER.debug( 

72 f"Sending request for {client_call_details.method}", 

73 extra={ 

74 "serviceName": "google.logging.v2.LoggingServiceV2", 

75 "rpcName": str(client_call_details.method), 

76 "request": grpc_request, 

77 "metadata": grpc_request["metadata"], 

78 }, 

79 ) 

80 response = continuation(client_call_details, request) 

81 if logging_enabled: # pragma: NO COVER 

82 response_metadata = response.trailing_metadata() 

83 # Convert gRPC metadata `<class 'grpc.aio._metadata.Metadata'>` to list of tuples 

84 metadata = ( 

85 dict([(k, str(v)) for k, v in response_metadata]) 

86 if response_metadata 

87 else None 

88 ) 

89 result = response.result() 

90 if isinstance(result, proto.Message): 

91 response_payload = type(result).to_json(result) 

92 elif isinstance(result, google.protobuf.message.Message): 

93 response_payload = MessageToJson(result) 

94 else: 

95 response_payload = f"{type(result).__name__}: {pickle.dumps(result)}" 

96 grpc_response = { 

97 "payload": response_payload, 

98 "metadata": metadata, 

99 "status": "OK", 

100 } 

101 _LOGGER.debug( 

102 f"Received response for {client_call_details.method}.", 

103 extra={ 

104 "serviceName": "google.logging.v2.LoggingServiceV2", 

105 "rpcName": client_call_details.method, 

106 "response": grpc_response, 

107 "metadata": grpc_response["metadata"], 

108 }, 

109 ) 

110 return response 

111 

112 

113class LoggingServiceV2GrpcTransport(LoggingServiceV2Transport): 

114 """gRPC backend transport for LoggingServiceV2. 

115 

116 Service for ingesting and querying logs. 

117 

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

119 primary client can load the underlying transport implementation 

120 and call it. 

121 

122 It sends protocol buffers over the wire using gRPC (which is built on 

123 top of HTTP/2); the ``grpcio`` package must be installed. 

124 """ 

125 

126 _stubs: Dict[str, Callable] 

127 

128 def __init__( 

129 self, 

130 *, 

131 host: str = "logging.googleapis.com", 

132 credentials: Optional[ga_credentials.Credentials] = None, 

133 credentials_file: Optional[str] = None, 

134 scopes: Optional[Sequence[str]] = None, 

135 channel: Optional[Union[grpc.Channel, Callable[..., grpc.Channel]]] = None, 

136 api_mtls_endpoint: Optional[str] = None, 

137 client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]] = None, 

138 ssl_channel_credentials: Optional[grpc.ChannelCredentials] = None, 

139 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None, 

140 quota_project_id: Optional[str] = None, 

141 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

142 always_use_jwt_access: Optional[bool] = False, 

143 api_audience: Optional[str] = None, 

144 ) -> None: 

145 """Instantiate the transport. 

146 

147 Args: 

148 host (Optional[str]): 

149 The hostname to connect to (default: 'logging.googleapis.com'). 

150 credentials (Optional[google.auth.credentials.Credentials]): The 

151 authorization credentials to attach to requests. These 

152 credentials identify the application to the service; if none 

153 are specified, the client will attempt to ascertain the 

154 credentials from the environment. 

155 This argument is ignored if a ``channel`` instance is provided. 

156 credentials_file (Optional[str]): Deprecated. A file with credentials that can 

157 be loaded with :func:`google.auth.load_credentials_from_file`. 

158 This argument is ignored if a ``channel`` instance is provided. 

159 This argument will be removed in the next major version of this library. 

160 scopes (Optional(Sequence[str])): A list of scopes. This argument is 

161 ignored if a ``channel`` instance is provided. 

162 channel (Optional[Union[grpc.Channel, Callable[..., grpc.Channel]]]): 

163 A ``Channel`` instance through which to make calls, or a Callable 

164 that constructs and returns one. If set to None, ``self.create_channel`` 

165 is used to create the channel. If a Callable is given, it will be called 

166 with the same arguments as used in ``self.create_channel``. 

167 api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint. 

168 If provided, it overrides the ``host`` argument and tries to create 

169 a mutual TLS channel with client SSL credentials from 

170 ``client_cert_source`` or application default SSL credentials. 

171 client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]): 

172 Deprecated. A callback to provide client SSL certificate bytes and 

173 private key bytes, both in PEM format. It is ignored if 

174 ``api_mtls_endpoint`` is None. 

175 ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials 

176 for the grpc channel. It is ignored if a ``channel`` instance is provided. 

177 client_cert_source_for_mtls (Optional[Callable[[], Tuple[bytes, bytes]]]): 

178 A callback to provide client certificate bytes and private key bytes, 

179 both in PEM format. It is used to configure a mutual TLS channel. It is 

180 ignored if a ``channel`` instance or ``ssl_channel_credentials`` is provided. 

181 quota_project_id (Optional[str]): An optional project to use for billing 

182 and quota. 

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

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

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

186 Generally, you only need to set this if you're developing 

187 your own client library. 

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

189 be used for service account credentials. 

190 

191 Raises: 

192 google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport 

193 creation failed for any reason. 

194 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials`` 

195 and ``credentials_file`` are passed. 

196 """ 

197 self._grpc_channel = None 

198 self._ssl_channel_credentials = ssl_channel_credentials 

199 self._stubs: Dict[str, Callable] = {} 

200 

201 if api_mtls_endpoint: 

202 warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning) 

203 if client_cert_source: 

204 warnings.warn("client_cert_source is deprecated", DeprecationWarning) 

205 

206 if isinstance(channel, grpc.Channel): 

207 # Ignore credentials if a channel was passed. 

208 credentials = None 

209 self._ignore_credentials = True 

210 # If a channel was explicitly provided, set it. 

211 self._grpc_channel = channel 

212 self._ssl_channel_credentials = None 

213 

214 else: 

215 if api_mtls_endpoint: 

216 host = api_mtls_endpoint 

217 

218 # Create SSL credentials with client_cert_source or application 

219 # default SSL credentials. 

220 if client_cert_source: 

221 cert, key = client_cert_source() 

222 self._ssl_channel_credentials = grpc.ssl_channel_credentials( 

223 certificate_chain=cert, private_key=key 

224 ) 

225 else: 

226 self._ssl_channel_credentials = SslCredentials().ssl_credentials 

227 

228 else: 

229 if client_cert_source_for_mtls and not ssl_channel_credentials: 

230 cert, key = client_cert_source_for_mtls() 

231 self._ssl_channel_credentials = grpc.ssl_channel_credentials( 

232 certificate_chain=cert, private_key=key 

233 ) 

234 

235 # The base transport sets the host, credentials and scopes 

236 super().__init__( 

237 host=host, 

238 credentials=credentials, 

239 credentials_file=credentials_file, 

240 scopes=scopes, 

241 quota_project_id=quota_project_id, 

242 client_info=client_info, 

243 always_use_jwt_access=always_use_jwt_access, 

244 api_audience=api_audience, 

245 ) 

246 

247 if not self._grpc_channel: 

248 # initialize with the provided callable or the default channel 

249 channel_init = channel or type(self).create_channel 

250 self._grpc_channel = channel_init( 

251 self._host, 

252 # use the credentials which are saved 

253 credentials=self._credentials, 

254 # Set ``credentials_file`` to ``None`` here as 

255 # the credentials that we saved earlier should be used. 

256 credentials_file=None, 

257 scopes=self._scopes, 

258 ssl_credentials=self._ssl_channel_credentials, 

259 quota_project_id=quota_project_id, 

260 options=[ 

261 ("grpc.max_send_message_length", -1), 

262 ("grpc.max_receive_message_length", -1), 

263 ], 

264 ) 

265 

266 self._interceptor = _LoggingClientInterceptor() 

267 self._logged_channel = grpc.intercept_channel( 

268 self._grpc_channel, self._interceptor 

269 ) 

270 

271 # Wrap messages. This must be done after self._logged_channel exists 

272 self._prep_wrapped_messages(client_info) 

273 

274 @classmethod 

275 def create_channel( 

276 cls, 

277 host: str = "logging.googleapis.com", 

278 credentials: Optional[ga_credentials.Credentials] = None, 

279 credentials_file: Optional[str] = None, 

280 scopes: Optional[Sequence[str]] = None, 

281 quota_project_id: Optional[str] = None, 

282 **kwargs, 

283 ) -> grpc.Channel: 

284 """Create and return a gRPC channel object. 

285 Args: 

286 host (Optional[str]): The host for the channel to use. 

287 credentials (Optional[~.Credentials]): The 

288 authorization credentials to attach to requests. These 

289 credentials identify this application to the service. If 

290 none are specified, the client will attempt to ascertain 

291 the credentials from the environment. 

292 credentials_file (Optional[str]): Deprecated. A file with credentials that can 

293 be loaded with :func:`google.auth.load_credentials_from_file`. 

294 This argument is mutually exclusive with credentials. This argument will be 

295 removed in the next major version of this library. 

296 scopes (Optional[Sequence[str]]): A optional list of scopes needed for this 

297 service. These are only used when credentials are not specified and 

298 are passed to :func:`google.auth.default`. 

299 quota_project_id (Optional[str]): An optional project to use for billing 

300 and quota. 

301 kwargs (Optional[dict]): Keyword arguments, which are passed to the 

302 channel creation. 

303 Returns: 

304 grpc.Channel: A gRPC channel object. 

305 

306 Raises: 

307 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials`` 

308 and ``credentials_file`` are passed. 

309 """ 

310 

311 return grpc_helpers.create_channel( 

312 host, 

313 credentials=credentials, 

314 credentials_file=credentials_file, 

315 quota_project_id=quota_project_id, 

316 default_scopes=cls.AUTH_SCOPES, 

317 scopes=scopes, 

318 default_host=cls.DEFAULT_HOST, 

319 **kwargs, 

320 ) 

321 

322 @property 

323 def grpc_channel(self) -> grpc.Channel: 

324 """Return the channel designed to connect to this service.""" 

325 return self._grpc_channel 

326 

327 @property 

328 def delete_log(self) -> Callable[[logging.DeleteLogRequest], empty_pb2.Empty]: 

329 r"""Return a callable for the delete log method over gRPC. 

330 

331 Deletes all the log entries in a log for the \_Default Log 

332 Bucket. The log reappears if it receives new entries. Log 

333 entries written shortly before the delete operation might not be 

334 deleted. Entries received after the delete operation with a 

335 timestamp before the operation will be deleted. 

336 

337 Returns: 

338 Callable[[~.DeleteLogRequest], 

339 ~.Empty]: 

340 A function that, when called, will call the underlying RPC 

341 on the server. 

342 """ 

343 # Generate a "stub function" on-the-fly which will actually make 

344 # the request. 

345 # gRPC handles serialization and deserialization, so we just need 

346 # to pass in the functions for each. 

347 if "delete_log" not in self._stubs: 

348 self._stubs["delete_log"] = self._logged_channel.unary_unary( 

349 "/google.logging.v2.LoggingServiceV2/DeleteLog", 

350 request_serializer=logging.DeleteLogRequest.serialize, 

351 response_deserializer=empty_pb2.Empty.FromString, 

352 ) 

353 return self._stubs["delete_log"] 

354 

355 @property 

356 def write_log_entries( 

357 self, 

358 ) -> Callable[[logging.WriteLogEntriesRequest], logging.WriteLogEntriesResponse]: 

359 r"""Return a callable for the write log entries method over gRPC. 

360 

361 Writes log entries to Logging. This API method is the 

362 only way to send log entries to Logging. This method is 

363 used, directly or indirectly, by the Logging agent 

364 (fluentd) and all logging libraries configured to use 

365 Logging. A single request may contain log entries for a 

366 maximum of 1000 different resources (projects, 

367 organizations, billing accounts or folders) 

368 

369 Returns: 

370 Callable[[~.WriteLogEntriesRequest], 

371 ~.WriteLogEntriesResponse]: 

372 A function that, when called, will call the underlying RPC 

373 on the server. 

374 """ 

375 # Generate a "stub function" on-the-fly which will actually make 

376 # the request. 

377 # gRPC handles serialization and deserialization, so we just need 

378 # to pass in the functions for each. 

379 if "write_log_entries" not in self._stubs: 

380 self._stubs["write_log_entries"] = self._logged_channel.unary_unary( 

381 "/google.logging.v2.LoggingServiceV2/WriteLogEntries", 

382 request_serializer=logging.WriteLogEntriesRequest.serialize, 

383 response_deserializer=logging.WriteLogEntriesResponse.deserialize, 

384 ) 

385 return self._stubs["write_log_entries"] 

386 

387 @property 

388 def list_log_entries( 

389 self, 

390 ) -> Callable[[logging.ListLogEntriesRequest], logging.ListLogEntriesResponse]: 

391 r"""Return a callable for the list log entries method over gRPC. 

392 

393 Lists log entries. Use this method to retrieve log entries that 

394 originated from a project/folder/organization/billing account. 

395 For ways to export log entries, see `Exporting 

396 Logs <https://cloud.google.com/logging/docs/export>`__. 

397 

398 Returns: 

399 Callable[[~.ListLogEntriesRequest], 

400 ~.ListLogEntriesResponse]: 

401 A function that, when called, will call the underlying RPC 

402 on the server. 

403 """ 

404 # Generate a "stub function" on-the-fly which will actually make 

405 # the request. 

406 # gRPC handles serialization and deserialization, so we just need 

407 # to pass in the functions for each. 

408 if "list_log_entries" not in self._stubs: 

409 self._stubs["list_log_entries"] = self._logged_channel.unary_unary( 

410 "/google.logging.v2.LoggingServiceV2/ListLogEntries", 

411 request_serializer=logging.ListLogEntriesRequest.serialize, 

412 response_deserializer=logging.ListLogEntriesResponse.deserialize, 

413 ) 

414 return self._stubs["list_log_entries"] 

415 

416 @property 

417 def list_monitored_resource_descriptors( 

418 self, 

419 ) -> Callable[ 

420 [logging.ListMonitoredResourceDescriptorsRequest], 

421 logging.ListMonitoredResourceDescriptorsResponse, 

422 ]: 

423 r"""Return a callable for the list monitored resource 

424 descriptors method over gRPC. 

425 

426 Lists the descriptors for monitored resource types 

427 used by Logging. 

428 

429 Returns: 

430 Callable[[~.ListMonitoredResourceDescriptorsRequest], 

431 ~.ListMonitoredResourceDescriptorsResponse]: 

432 A function that, when called, will call the underlying RPC 

433 on the server. 

434 """ 

435 # Generate a "stub function" on-the-fly which will actually make 

436 # the request. 

437 # gRPC handles serialization and deserialization, so we just need 

438 # to pass in the functions for each. 

439 if "list_monitored_resource_descriptors" not in self._stubs: 

440 self._stubs[ 

441 "list_monitored_resource_descriptors" 

442 ] = self._logged_channel.unary_unary( 

443 "/google.logging.v2.LoggingServiceV2/ListMonitoredResourceDescriptors", 

444 request_serializer=logging.ListMonitoredResourceDescriptorsRequest.serialize, 

445 response_deserializer=logging.ListMonitoredResourceDescriptorsResponse.deserialize, 

446 ) 

447 return self._stubs["list_monitored_resource_descriptors"] 

448 

449 @property 

450 def list_logs( 

451 self, 

452 ) -> Callable[[logging.ListLogsRequest], logging.ListLogsResponse]: 

453 r"""Return a callable for the list logs method over gRPC. 

454 

455 Lists the logs in projects, organizations, folders, 

456 or billing accounts. Only logs that have entries are 

457 listed. 

458 

459 Returns: 

460 Callable[[~.ListLogsRequest], 

461 ~.ListLogsResponse]: 

462 A function that, when called, will call the underlying RPC 

463 on the server. 

464 """ 

465 # Generate a "stub function" on-the-fly which will actually make 

466 # the request. 

467 # gRPC handles serialization and deserialization, so we just need 

468 # to pass in the functions for each. 

469 if "list_logs" not in self._stubs: 

470 self._stubs["list_logs"] = self._logged_channel.unary_unary( 

471 "/google.logging.v2.LoggingServiceV2/ListLogs", 

472 request_serializer=logging.ListLogsRequest.serialize, 

473 response_deserializer=logging.ListLogsResponse.deserialize, 

474 ) 

475 return self._stubs["list_logs"] 

476 

477 @property 

478 def tail_log_entries( 

479 self, 

480 ) -> Callable[[logging.TailLogEntriesRequest], logging.TailLogEntriesResponse]: 

481 r"""Return a callable for the tail log entries method over gRPC. 

482 

483 Streaming read of log entries as they are ingested. 

484 Until the stream is terminated, it will continue reading 

485 logs. 

486 

487 Returns: 

488 Callable[[~.TailLogEntriesRequest], 

489 ~.TailLogEntriesResponse]: 

490 A function that, when called, will call the underlying RPC 

491 on the server. 

492 """ 

493 # Generate a "stub function" on-the-fly which will actually make 

494 # the request. 

495 # gRPC handles serialization and deserialization, so we just need 

496 # to pass in the functions for each. 

497 if "tail_log_entries" not in self._stubs: 

498 self._stubs["tail_log_entries"] = self._logged_channel.stream_stream( 

499 "/google.logging.v2.LoggingServiceV2/TailLogEntries", 

500 request_serializer=logging.TailLogEntriesRequest.serialize, 

501 response_deserializer=logging.TailLogEntriesResponse.deserialize, 

502 ) 

503 return self._stubs["tail_log_entries"] 

504 

505 def close(self): 

506 self._logged_channel.close() 

507 

508 @property 

509 def cancel_operation( 

510 self, 

511 ) -> Callable[[operations_pb2.CancelOperationRequest], None]: 

512 r"""Return a callable for the cancel_operation method over gRPC.""" 

513 # Generate a "stub function" on-the-fly which will actually make 

514 # the request. 

515 # gRPC handles serialization and deserialization, so we just need 

516 # to pass in the functions for each. 

517 if "cancel_operation" not in self._stubs: 

518 self._stubs["cancel_operation"] = self._logged_channel.unary_unary( 

519 "/google.longrunning.Operations/CancelOperation", 

520 request_serializer=operations_pb2.CancelOperationRequest.SerializeToString, 

521 response_deserializer=None, 

522 ) 

523 return self._stubs["cancel_operation"] 

524 

525 @property 

526 def get_operation( 

527 self, 

528 ) -> Callable[[operations_pb2.GetOperationRequest], operations_pb2.Operation]: 

529 r"""Return a callable for the get_operation method over gRPC.""" 

530 # Generate a "stub function" on-the-fly which will actually make 

531 # the request. 

532 # gRPC handles serialization and deserialization, so we just need 

533 # to pass in the functions for each. 

534 if "get_operation" not in self._stubs: 

535 self._stubs["get_operation"] = self._logged_channel.unary_unary( 

536 "/google.longrunning.Operations/GetOperation", 

537 request_serializer=operations_pb2.GetOperationRequest.SerializeToString, 

538 response_deserializer=operations_pb2.Operation.FromString, 

539 ) 

540 return self._stubs["get_operation"] 

541 

542 @property 

543 def list_operations( 

544 self, 

545 ) -> Callable[ 

546 [operations_pb2.ListOperationsRequest], operations_pb2.ListOperationsResponse 

547 ]: 

548 r"""Return a callable for the list_operations method over gRPC.""" 

549 # Generate a "stub function" on-the-fly which will actually make 

550 # the request. 

551 # gRPC handles serialization and deserialization, so we just need 

552 # to pass in the functions for each. 

553 if "list_operations" not in self._stubs: 

554 self._stubs["list_operations"] = self._logged_channel.unary_unary( 

555 "/google.longrunning.Operations/ListOperations", 

556 request_serializer=operations_pb2.ListOperationsRequest.SerializeToString, 

557 response_deserializer=operations_pb2.ListOperationsResponse.FromString, 

558 ) 

559 return self._stubs["list_operations"] 

560 

561 @property 

562 def kind(self) -> str: 

563 return "grpc" 

564 

565 

566__all__ = ("LoggingServiceV2GrpcTransport",)