Waits for bq_job to finish running, up to a maximum amount of time specified by the timeout parameter (defaulting to 30 minutes). Args: client: A bigquery.client.Client to monitor the bq_job. bq_job: The bigquery.job.QueryJob that blocks until done runnning. timeout
(
client: Client,
bq_job: Union[bigquery.job.query.QueryJob, bigquery.job.load.LoadJob],
timeout: int = 1800,
retry_cadence: float = 1,
)
| 1320 | |
| 1321 | |
| 1322 | def block_until_done( |
| 1323 | client: Client, |
| 1324 | bq_job: Union[bigquery.job.query.QueryJob, bigquery.job.load.LoadJob], |
| 1325 | timeout: int = 1800, |
| 1326 | retry_cadence: float = 1, |
| 1327 | ): |
| 1328 | """ |
| 1329 | Waits for bq_job to finish running, up to a maximum amount of time specified by the timeout parameter (defaulting to 30 minutes). |
| 1330 | |
| 1331 | Args: |
| 1332 | client: A bigquery.client.Client to monitor the bq_job. |
| 1333 | bq_job: The bigquery.job.QueryJob that blocks until done runnning. |
| 1334 | timeout: An optional number of seconds for setting the time limit of the job. |
| 1335 | retry_cadence: An optional number of seconds for setting how long the job should checked for completion. |
| 1336 | |
| 1337 | Raises: |
| 1338 | BigQueryJobStillRunning exception if the function has blocked longer than 30 minutes. |
| 1339 | BigQueryJobCancelled exception to signify when that the job has been cancelled (i.e. from timeout or KeyboardInterrupt). |
| 1340 | """ |
| 1341 | |
| 1342 | # For test environments, retry more aggressively |
| 1343 | if flags_helper.is_test(): |
| 1344 | retry_cadence = 0.1 |
| 1345 | |
| 1346 | def _wait_until_done(bq_job): |
| 1347 | if client.get_job(bq_job).state in ["PENDING", "RUNNING"]: |
| 1348 | raise BigQueryJobStillRunning(job_id=bq_job.job_id) |
| 1349 | |
| 1350 | try: |
| 1351 | retryer = Retrying( |
| 1352 | wait=wait_fixed(retry_cadence), |
| 1353 | stop=stop_after_delay(timeout), |
| 1354 | retry=retry_if_exception_type(BigQueryJobStillRunning), |
| 1355 | reraise=True, |
| 1356 | ) |
| 1357 | retryer(_wait_until_done, bq_job) |
| 1358 | |
| 1359 | finally: |
| 1360 | if client.get_job(bq_job).state in ["PENDING", "RUNNING"]: |
| 1361 | client.cancel_job(bq_job.job_id) |
| 1362 | raise BigQueryJobCancelled(job_id=bq_job.job_id) |
| 1363 | |
| 1364 | # We explicitly set the timeout to None because `google-api-core` changed the default value and |
| 1365 | # breaks downstream libraries. |
| 1366 | # https://github.com/googleapis/python-api-core/issues/479 |
| 1367 | if bq_job.exception(timeout=None): |
| 1368 | raise bq_job.exception(timeout=None) |
| 1369 | |
| 1370 | |
| 1371 | def _get_table_reference_for_new_entity( |
no test coverage detected