Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/cloud/storage/notification.py: 39%

1# Copyright 2017 Google LLC 

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 

15"""Configure bucket notification resources to interact with Google Cloud Pub/Sub. 

16 

17See [Cloud Pub/Sub Notifications for Google Cloud Storage](https://cloud.google.com/storage/docs/pubsub-notifications) 

18""" 

19 

20import re 

21 

22from google.api_core.exceptions import NotFound 

23 

24from google.cloud.storage.constants import _DEFAULT_TIMEOUT 

25from google.cloud.storage.retry import DEFAULT_RETRY 

26 

27 

28OBJECT_FINALIZE_EVENT_TYPE = "OBJECT_FINALIZE" 

29OBJECT_METADATA_UPDATE_EVENT_TYPE = "OBJECT_METADATA_UPDATE" 

30OBJECT_DELETE_EVENT_TYPE = "OBJECT_DELETE" 

31OBJECT_ARCHIVE_EVENT_TYPE = "OBJECT_ARCHIVE" 

32 

33JSON_API_V1_PAYLOAD_FORMAT = "JSON_API_V1" 

34NONE_PAYLOAD_FORMAT = "NONE" 

35 

36_TOPIC_REF_FMT = "//pubsub.googleapis.com/projects/{}/topics/{}" 

37_PROJECT_PATTERN = r"(?P<project>[a-z][a-z0-9-]{4,28}[a-z0-9])" 

38_TOPIC_NAME_PATTERN = r"(?P<name>[A-Za-z](\w|[-_.~+%])+)" 

39_TOPIC_REF_PATTERN = _TOPIC_REF_FMT.format(_PROJECT_PATTERN, _TOPIC_NAME_PATTERN) 

40_TOPIC_REF_RE = re.compile(_TOPIC_REF_PATTERN) 

41_BAD_TOPIC = ( 

42 "Resource has invalid topic: {}; see " 

43 "https://cloud.google.com/storage/docs/json_api/v1/" 

44 "notifications/insert#topic" 

45) 

46 

47 

48class BucketNotification(object): 

49 """Represent a single notification resource for a bucket. 

50 

51 See: https://cloud.google.com/storage/docs/json_api/v1/notifications 

52 

53 :type bucket: :class:`google.cloud.storage.bucket.Bucket` 

54 :param bucket: Bucket to which the notification is bound. 

55 

56 :type topic_name: str 

57 :param topic_name: 

58 (Optional) Topic name to which notifications are published. 

59 

60 :type topic_project: str 

61 :param topic_project: 

62 (Optional) Project ID of topic to which notifications are published. 

63 If not passed, uses the project ID of the bucket's client. 

64 

65 :type custom_attributes: dict 

66 :param custom_attributes: 

67 (Optional) Additional attributes passed with notification events. 

68 

69 :type event_types: list(str) 

70 :param event_types: 

71 (Optional) Event types for which notification events are published. 

72 

73 :type blob_name_prefix: str 

74 :param blob_name_prefix: 

75 (Optional) Prefix of blob names for which notification events are 

76 published. 

77 

78 :type payload_format: str 

79 :param payload_format: 

80 (Optional) Format of payload for notification events. 

81 

82 :type notification_id: str 

83 :param notification_id: 

84 (Optional) The ID of the notification. 

85 """ 

86 

87 def __init__( 

88 self, 

89 bucket, 

90 topic_name=None, 

91 topic_project=None, 

92 custom_attributes=None, 

93 event_types=None, 

94 blob_name_prefix=None, 

95 payload_format=NONE_PAYLOAD_FORMAT, 

96 notification_id=None, 

97 ): 

98 self._bucket = bucket 

99 self._topic_name = topic_name 

100 

101 if topic_project is None: 

102 topic_project = bucket.client.project 

103 

104 if topic_project is None: 

105 raise ValueError("Client project not set: pass an explicit topic_project.") 

106 

107 self._topic_project = topic_project 

108 

109 self._properties = {} 

110 

111 if custom_attributes is not None: 

112 self._properties["custom_attributes"] = custom_attributes 

113 

114 if event_types is not None: 

115 self._properties["event_types"] = event_types 

116 

117 if blob_name_prefix is not None: 

118 self._properties["object_name_prefix"] = blob_name_prefix 

119 

120 if notification_id is not None: 

121 self._properties["id"] = notification_id 

122 

123 self._properties["payload_format"] = payload_format 

124 

125 @classmethod 

126 def from_api_repr(cls, resource, bucket): 

127 """Construct an instance from the JSON repr returned by the server. 

128 

129 See: https://cloud.google.com/storage/docs/json_api/v1/notifications 

130 

131 :type resource: dict 

132 :param resource: JSON repr of the notification 

133 

134 :type bucket: :class:`google.cloud.storage.bucket.Bucket` 

135 :param bucket: Bucket to which the notification is bound. 

136 

137 :rtype: :class:`BucketNotification` 

138 :returns: the new notification instance 

139 """ 

140 topic_path = resource.get("topic") 

141 if topic_path is None: 

142 raise ValueError("Resource has no topic") 

143 

144 name, project = _parse_topic_path(topic_path) 

145 instance = cls(bucket, name, topic_project=project) 

146 instance._properties = resource 

147 

148 return instance 

149 

150 @property 

151 def bucket(self): 

152 """Bucket to which the notification is bound.""" 

153 return self._bucket 

154 

155 @property 

156 def topic_name(self): 

157 """Topic name to which notifications are published.""" 

158 return self._topic_name 

159 

160 @property 

161 def topic_project(self): 

162 """Project ID of topic to which notifications are published.""" 

163 return self._topic_project 

164 

165 @property 

166 def custom_attributes(self): 

167 """Custom attributes passed with notification events.""" 

168 return self._properties.get("custom_attributes") 

169 

170 @property 

171 def event_types(self): 

172 """Event types for which notification events are published.""" 

173 return self._properties.get("event_types") 

174 

175 @property 

176 def blob_name_prefix(self): 

177 """Prefix of blob names for which notification events are published.""" 

178 return self._properties.get("object_name_prefix") 

179 

180 @property 

181 def payload_format(self): 

182 """Format of payload of notification events.""" 

183 return self._properties.get("payload_format") 

184 

185 @property 

186 def notification_id(self): 

187 """Server-set ID of notification resource.""" 

188 return self._properties.get("id") 

189 

190 @property 

191 def etag(self): 

192 """Server-set ETag of notification resource.""" 

193 return self._properties.get("etag") 

194 

195 @property 

196 def self_link(self): 

197 """Server-set ETag of notification resource.""" 

198 return self._properties.get("selfLink") 

199 

200 @property 

201 def client(self): 

202 """The client bound to this notfication.""" 

203 return self.bucket.client 

204 

205 @property 

206 def path(self): 

207 """The URL path for this notification.""" 

208 return f"/b/{self.bucket.name}/notificationConfigs/{self.notification_id}" 

209 

210 def _require_client(self, client): 

211 """Check client or verify over-ride. 

212 

213 :type client: :class:`~google.cloud.storage.client.Client` or 

214 ``NoneType`` 

215 :param client: the client to use. 

216 

217 :rtype: :class:`google.cloud.storage.client.Client` 

218 :returns: The client passed in or the bucket's client. 

219 """ 

220 if client is None: 

221 client = self.client 

222 return client 

223 

224 def _set_properties(self, response): 

225 """Helper for :meth:`reload`. 

226 

227 :type response: dict 

228 :param response: resource mapping from server 

229 """ 

230 self._properties.clear() 

231 self._properties.update(response) 

232 

233 def create(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=None): 

234 """API wrapper: create the notification. 

235 

236 See: 

237 https://cloud.google.com/storage/docs/json_api/v1/notifications/insert 

238 

239 If :attr:`user_project` is set on the bucket, bills the API request 

240 to that project. 

241 

242 :type client: :class:`~google.cloud.storage.client.Client` 

243 :param client: (Optional) The client to use. If not passed, falls back 

244 to the ``client`` stored on the notification's bucket. 

245 :type timeout: float or tuple 

246 :param timeout: 

247 (Optional) The amount of time, in seconds, to wait 

248 for the server response. See: :ref:`configuring_timeouts` 

249 

250 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

251 :param retry: 

252 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

253 

254 :raises ValueError: if the notification already exists. 

255 """ 

256 if self.notification_id is not None: 

257 raise ValueError( 

258 f"Notification already exists w/ id: {self.notification_id}" 

259 ) 

260 

261 client = self._require_client(client) 

262 

263 query_params = {} 

264 if self.bucket.user_project is not None: 

265 query_params["userProject"] = self.bucket.user_project 

266 

267 path = f"/b/{self.bucket.name}/notificationConfigs" 

268 properties = self._properties.copy() 

269 

270 if self.topic_name is None: 

271 properties["topic"] = _TOPIC_REF_FMT.format(self.topic_project, "") 

272 else: 

273 properties["topic"] = _TOPIC_REF_FMT.format( 

274 self.topic_project, self.topic_name 

275 ) 

276 

277 self._properties = client._post_resource( 

278 path, 

279 properties, 

280 query_params=query_params, 

281 timeout=timeout, 

282 retry=retry, 

283 ) 

284 

285 def exists(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY): 

286 """Test whether this notification exists. 

287 

288 See: 

289 https://cloud.google.com/storage/docs/json_api/v1/notifications/get 

290 

291 If :attr:`user_project` is set on the bucket, bills the API request 

292 to that project. 

293 

294 :type client: :class:`~google.cloud.storage.client.Client` or 

295 ``NoneType`` 

296 :param client: (Optional) The client to use. If not passed, falls back 

297 to the ``client`` stored on the current bucket. 

298 :type timeout: float or tuple 

299 :param timeout: 

300 (Optional) The amount of time, in seconds, to wait 

301 for the server response. See: :ref:`configuring_timeouts` 

302 

303 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

304 :param retry: 

305 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

306 

307 :rtype: bool 

308 :returns: True, if the notification exists, else False. 

309 :raises ValueError: if the notification has no ID. 

310 """ 

311 if self.notification_id is None: 

312 raise ValueError("Notification ID not set: set an explicit notification_id") 

313 

314 client = self._require_client(client) 

315 

316 query_params = {} 

317 if self.bucket.user_project is not None: 

318 query_params["userProject"] = self.bucket.user_project 

319 

320 try: 

321 client._get_resource( 

322 self.path, 

323 query_params=query_params, 

324 timeout=timeout, 

325 retry=retry, 

326 ) 

327 except NotFound: 

328 return False 

329 else: 

330 return True 

331 

332 def reload(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY): 

333 """Update this notification from the server configuration. 

334 

335 See: 

336 https://cloud.google.com/storage/docs/json_api/v1/notifications/get 

337 

338 If :attr:`user_project` is set on the bucket, bills the API request 

339 to that project. 

340 

341 :type client: :class:`~google.cloud.storage.client.Client` or 

342 ``NoneType`` 

343 :param client: (Optional) The client to use. If not passed, falls back 

344 to the ``client`` stored on the current bucket. 

345 :type timeout: float or tuple 

346 :param timeout: 

347 (Optional) The amount of time, in seconds, to wait 

348 for the server response. See: :ref:`configuring_timeouts` 

349 

350 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

351 :param retry: 

352 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

353 

354 

355 :raises ValueError: if the notification has no ID. 

356 """ 

357 if self.notification_id is None: 

358 raise ValueError("Notification ID not set: set an explicit notification_id") 

359 

360 client = self._require_client(client) 

361 

362 query_params = {} 

363 if self.bucket.user_project is not None: 

364 query_params["userProject"] = self.bucket.user_project 

365 

366 response = client._get_resource( 

367 self.path, 

368 query_params=query_params, 

369 timeout=timeout, 

370 retry=retry, 

371 ) 

372 self._set_properties(response) 

373 

374 def delete(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY): 

375 """Delete this notification. 

376 

377 See: 

378 https://cloud.google.com/storage/docs/json_api/v1/notifications/delete 

379 

380 If :attr:`user_project` is set on the bucket, bills the API request 

381 to that project. 

382 

383 :type client: :class:`~google.cloud.storage.client.Client` or 

384 ``NoneType`` 

385 :param client: (Optional) The client to use. If not passed, falls back 

386 to the ``client`` stored on the current bucket. 

387 :type timeout: float or tuple 

388 :param timeout: 

389 (Optional) The amount of time, in seconds, to wait 

390 for the server response. See: :ref:`configuring_timeouts` 

391 

392 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

393 :param retry: 

394 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

395 

396 :raises: :class:`google.api_core.exceptions.NotFound`: 

397 if the notification does not exist. 

398 :raises ValueError: if the notification has no ID. 

399 """ 

400 if self.notification_id is None: 

401 raise ValueError("Notification ID not set: set an explicit notification_id") 

402 

403 client = self._require_client(client) 

404 

405 query_params = {} 

406 if self.bucket.user_project is not None: 

407 query_params["userProject"] = self.bucket.user_project 

408 

409 client._delete_resource( 

410 self.path, 

411 query_params=query_params, 

412 timeout=timeout, 

413 retry=retry, 

414 ) 

415 

416 

417def _parse_topic_path(topic_path): 

418 """Verify that a topic path is in the correct format. 

419 

420 Expected to be of the form: 

421 

422 //pubsub.googleapis.com/projects/{project}/topics/{topic} 

423 

424 where the ``project`` value must be "6 to 30 lowercase letters, digits, 

425 or hyphens. It must start with a letter. Trailing hyphens are prohibited." 

426 (see [`resource manager docs`](https://cloud.google.com/resource-manager/reference/rest/v1beta1/projects#Project.FIELDS.project_id)) 

427 and ``topic`` must have length at least two, 

428 must start with a letter and may only contain alphanumeric characters or 

429 ``-``, ``_``, ``.``, ``~``, ``+`` or ``%`` (i.e characters used for URL 

430 encoding, see [`topic spec`](https://cloud.google.com/storage/docs/json_api/v1/notifications/insert#topic)). 

431 

432 Args: 

433 topic_path (str): The topic path to be verified. 

434 

435 Returns: 

436 Tuple[str, str]: The ``project`` and ``topic`` parsed from the 

437 ``topic_path``. 

438 

439 Raises: 

440 ValueError: If the topic path is invalid. 

441 """ 

442 match = _TOPIC_REF_RE.match(topic_path) 

443 if match is None: 

444 raise ValueError(_BAD_TOPIC.format(topic_path)) 

445 

446 return match.group("name"), match.group("project")