创建错误条件：google.api_core.exceptions.NotFound: 404 Not found: URI答案

【问题标题】：Create a condition on error: google.api_core.exceptions.NotFound: 404 Not found: URI创建错误条件：google.api_core.exceptions.NotFound: 404 Not found: URI
【发布时间】：2019-12-09 20:13:13
【问题描述】：

我想知道如果在云存储中没有找到csv文件，是否可以创建条件，终止脚本（例如打印）。

def insert_bigquery(target_uri, dataset_id, table_id):
    bigquery_client = bigquery.Client()
    dataset_ref = bigquery_client.dataset(dataset_id)
    job_config = bigquery.LoadJobConfig()
    job_config.schema = [
        bigquery.SchemaField('id','STRING',mode='REQUIRED')
    ]
    job_config.source_format = bigquery.SourceFormat.CSV
    job_config.field_delimiter = ";"
    uri = target_uri
    load_job = bigquery_client.load_table_from_uri(
        uri,
        dataset_ref.table(table_id),
        job_config=job_config
        )
    print('Starting job {}'.format(load_job.job_id))
    load_job.result()
    print('Job finished.')

insert_bigquery(bucket_name, dataset, tabela)

【问题讨论】：

这个答案可能会有所帮助 - stackoverflow.com/questions/13525482/…
@PriyaAgarwal 这是相关的，但由于我不知道即将到来的 csv 的确切名称，我可以在目录级别进行验证吗？或者有什么方法可以捕获即将到来的 csv 的名称？

标签： python google-bigquery

【解决方案1】：

由于我不知道会出现什么 csv，所以我做了一个目录级别的测试，它成功了。

def blob_exists():
    client = storage.Client()
    bucket = client.get_bucket('bucket_name')
    bb = bucket.list_blobs(prefix='test_directory')
    for resource in bb:
        print(resource.name)
        blob = bucket.blob(resource.name)
        return blob.exists()

a = blob_exists()

if a is True:
    print("OK")
else:
    print("Not found")

【讨论】：