Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/cloud/bigquery_storage_v1/services/big_query_read/transports/grpc.py: 43%

65 statements  

« prev     ^ index     » next       coverage.py v7.2.2, created at 2023-03-26 06:10 +0000

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

2# Copyright 2022 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 warnings 

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

18 

19from google.api_core import grpc_helpers 

20from google.api_core import gapic_v1 

21import google.auth # type: ignore 

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

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

24 

25import grpc # type: ignore 

26 

27from google.cloud.bigquery_storage_v1.types import storage 

28from google.cloud.bigquery_storage_v1.types import stream 

29from .base import BigQueryReadTransport, DEFAULT_CLIENT_INFO 

30 

31 

32class BigQueryReadGrpcTransport(BigQueryReadTransport): 

33 """gRPC backend transport for BigQueryRead. 

34 

35 BigQuery Read API. 

36 The Read API can be used to read data from BigQuery. 

37 

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

39 primary client can load the underlying transport implementation 

40 and call it. 

41 

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

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

44 """ 

45 

46 _stubs: Dict[str, Callable] 

47 

48 def __init__( 

49 self, 

50 *, 

51 host: str = "bigquerystorage.googleapis.com", 

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

53 credentials_file: Optional[str] = None, 

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

55 channel: Optional[grpc.Channel] = None, 

56 api_mtls_endpoint: Optional[str] = None, 

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

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

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

60 quota_project_id: Optional[str] = None, 

61 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

62 always_use_jwt_access: Optional[bool] = False, 

63 api_audience: Optional[str] = None, 

64 ) -> None: 

65 """Instantiate the transport. 

66 

67 Args: 

68 host (Optional[str]): 

69 The hostname to connect to. 

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

71 authorization credentials to attach to requests. These 

72 credentials identify the application to the service; if none 

73 are specified, the client will attempt to ascertain the 

74 credentials from the environment. 

75 This argument is ignored if ``channel`` is provided. 

76 credentials_file (Optional[str]): A file with credentials that can 

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

78 This argument is ignored if ``channel`` is provided. 

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

80 ignored if ``channel`` is provided. 

81 channel (Optional[grpc.Channel]): A ``Channel`` instance through 

82 which to make calls. 

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

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

85 a mutual TLS channel with client SSL credentials from 

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

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

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

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

90 ``api_mtls_endpoint`` is None. 

91 ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials 

92 for the grpc channel. It is ignored if ``channel`` is provided. 

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

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

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

96 ignored if ``channel`` or ``ssl_channel_credentials`` is provided. 

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

98 and quota. 

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

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

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

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

103 your own client library. 

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

105 be used for service account credentials. 

106 

107 Raises: 

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

109 creation failed for any reason. 

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

111 and ``credentials_file`` are passed. 

112 """ 

113 self._grpc_channel = None 

114 self._ssl_channel_credentials = ssl_channel_credentials 

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

116 

117 if api_mtls_endpoint: 

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

119 if client_cert_source: 

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

121 

122 if channel: 

123 # Ignore credentials if a channel was passed. 

124 credentials = False 

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

126 self._grpc_channel = channel 

127 self._ssl_channel_credentials = None 

128 

129 else: 

130 if api_mtls_endpoint: 

131 host = api_mtls_endpoint 

132 

133 # Create SSL credentials with client_cert_source or application 

134 # default SSL credentials. 

135 if client_cert_source: 

136 cert, key = client_cert_source() 

137 self._ssl_channel_credentials = grpc.ssl_channel_credentials( 

138 certificate_chain=cert, private_key=key 

139 ) 

140 else: 

141 self._ssl_channel_credentials = SslCredentials().ssl_credentials 

142 

143 else: 

144 if client_cert_source_for_mtls and not ssl_channel_credentials: 

145 cert, key = client_cert_source_for_mtls() 

146 self._ssl_channel_credentials = grpc.ssl_channel_credentials( 

147 certificate_chain=cert, private_key=key 

148 ) 

149 

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

151 super().__init__( 

152 host=host, 

153 credentials=credentials, 

154 credentials_file=credentials_file, 

155 scopes=scopes, 

156 quota_project_id=quota_project_id, 

157 client_info=client_info, 

158 always_use_jwt_access=always_use_jwt_access, 

159 api_audience=api_audience, 

160 ) 

161 

162 if not self._grpc_channel: 

163 self._grpc_channel = type(self).create_channel( 

164 self._host, 

165 # use the credentials which are saved 

166 credentials=self._credentials, 

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

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

169 credentials_file=None, 

170 scopes=self._scopes, 

171 ssl_credentials=self._ssl_channel_credentials, 

172 quota_project_id=quota_project_id, 

173 options=[ 

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

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

176 ], 

177 ) 

178 

179 # Wrap messages. This must be done after self._grpc_channel exists 

180 self._prep_wrapped_messages(client_info) 

181 

182 @classmethod 

183 def create_channel( 

184 cls, 

185 host: str = "bigquerystorage.googleapis.com", 

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

187 credentials_file: Optional[str] = None, 

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

189 quota_project_id: Optional[str] = None, 

190 **kwargs, 

191 ) -> grpc.Channel: 

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

193 Args: 

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

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

196 authorization credentials to attach to requests. These 

197 credentials identify this application to the service. If 

198 none are specified, the client will attempt to ascertain 

199 the credentials from the environment. 

200 credentials_file (Optional[str]): A file with credentials that can 

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

202 This argument is mutually exclusive with credentials. 

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

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

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

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

207 and quota. 

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

209 channel creation. 

210 Returns: 

211 grpc.Channel: A gRPC channel object. 

212 

213 Raises: 

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

215 and ``credentials_file`` are passed. 

216 """ 

217 

218 return grpc_helpers.create_channel( 

219 host, 

220 credentials=credentials, 

221 credentials_file=credentials_file, 

222 quota_project_id=quota_project_id, 

223 default_scopes=cls.AUTH_SCOPES, 

224 scopes=scopes, 

225 default_host=cls.DEFAULT_HOST, 

226 **kwargs, 

227 ) 

228 

229 @property 

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

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

232 return self._grpc_channel 

233 

234 @property 

235 def create_read_session( 

236 self, 

237 ) -> Callable[[storage.CreateReadSessionRequest], stream.ReadSession]: 

238 r"""Return a callable for the create read session method over gRPC. 

239 

240 Creates a new read session. A read session divides 

241 the contents of a BigQuery table into one or more 

242 streams, which can then be used to read data from the 

243 table. The read session also specifies properties of the 

244 data to be read, such as a list of columns or a 

245 push-down filter describing the rows to be returned. 

246 

247 A particular row can be read by at most one stream. When 

248 the caller has reached the end of each stream in the 

249 session, then all the data in the table has been read. 

250 

251 Data is assigned to each stream such that roughly the 

252 same number of rows can be read from each stream. 

253 Because the server-side unit for assigning data is 

254 collections of rows, the API does not guarantee that 

255 each stream will return the same number or rows. 

256 Additionally, the limits are enforced based on the 

257 number of pre-filtered rows, so some filters can lead to 

258 lopsided assignments. 

259 

260 Read sessions automatically expire 6 hours after they 

261 are created and do not require manual clean-up by the 

262 caller. 

263 

264 Returns: 

265 Callable[[~.CreateReadSessionRequest], 

266 ~.ReadSession]: 

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

268 on the server. 

269 """ 

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

271 # the request. 

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

273 # to pass in the functions for each. 

274 if "create_read_session" not in self._stubs: 

275 self._stubs["create_read_session"] = self.grpc_channel.unary_unary( 

276 "/google.cloud.bigquery.storage.v1.BigQueryRead/CreateReadSession", 

277 request_serializer=storage.CreateReadSessionRequest.serialize, 

278 response_deserializer=stream.ReadSession.deserialize, 

279 ) 

280 return self._stubs["create_read_session"] 

281 

282 @property 

283 def read_rows( 

284 self, 

285 ) -> Callable[[storage.ReadRowsRequest], storage.ReadRowsResponse]: 

286 r"""Return a callable for the read rows method over gRPC. 

287 

288 Reads rows from the stream in the format prescribed 

289 by the ReadSession. Each response contains one or more 

290 table rows, up to a maximum of 100 MiB per response; 

291 read requests which attempt to read individual rows 

292 larger than 100 MiB will fail. 

293 

294 Each request also returns a set of stream statistics 

295 reflecting the current state of the stream. 

296 

297 Returns: 

298 Callable[[~.ReadRowsRequest], 

299 ~.ReadRowsResponse]: 

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

301 on the server. 

302 """ 

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

304 # the request. 

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

306 # to pass in the functions for each. 

307 if "read_rows" not in self._stubs: 

308 self._stubs["read_rows"] = self.grpc_channel.unary_stream( 

309 "/google.cloud.bigquery.storage.v1.BigQueryRead/ReadRows", 

310 request_serializer=storage.ReadRowsRequest.serialize, 

311 response_deserializer=storage.ReadRowsResponse.deserialize, 

312 ) 

313 return self._stubs["read_rows"] 

314 

315 @property 

316 def split_read_stream( 

317 self, 

318 ) -> Callable[[storage.SplitReadStreamRequest], storage.SplitReadStreamResponse]: 

319 r"""Return a callable for the split read stream method over gRPC. 

320 

321 Splits a given ``ReadStream`` into two ``ReadStream`` objects. 

322 These ``ReadStream`` objects are referred to as the primary and 

323 the residual streams of the split. The original ``ReadStream`` 

324 can still be read from in the same manner as before. Both of the 

325 returned ``ReadStream`` objects can also be read from, and the 

326 rows returned by both child streams will be the same as the rows 

327 read from the original stream. 

328 

329 Moreover, the two child streams will be allocated back-to-back 

330 in the original ``ReadStream``. Concretely, it is guaranteed 

331 that for streams original, primary, and residual, that 

332 original[0-j] = primary[0-j] and original[j-n] = residual[0-m] 

333 once the streams have been read to completion. 

334 

335 Returns: 

336 Callable[[~.SplitReadStreamRequest], 

337 ~.SplitReadStreamResponse]: 

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

339 on the server. 

340 """ 

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

342 # the request. 

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

344 # to pass in the functions for each. 

345 if "split_read_stream" not in self._stubs: 

346 self._stubs["split_read_stream"] = self.grpc_channel.unary_unary( 

347 "/google.cloud.bigquery.storage.v1.BigQueryRead/SplitReadStream", 

348 request_serializer=storage.SplitReadStreamRequest.serialize, 

349 response_deserializer=storage.SplitReadStreamResponse.deserialize, 

350 ) 

351 return self._stubs["split_read_stream"] 

352 

353 def close(self): 

354 self.grpc_channel.close() 

355 

356 @property 

357 def kind(self) -> str: 

358 return "grpc" 

359 

360 

361__all__ = ("BigQueryReadGrpcTransport",)