【问题标题】:How to check if file exists in Google Cloud Storage?如何检查文件是否存在于 Google Cloud Storage 中?
【发布时间】:2012-11-11 14:32:46
【问题描述】:

我有一个脚本,我想检查存储桶中是否存在文件,如果不存在则创建一个。

我尝试使用os.path.exists(file_path) where file_path = "/gs/testbucket",但我得到了一个找不到文件的错误。

我知道我可以使用files.listdir() API 函数列出位于路径中的所有文件,然后检查我想要的文件是否是其中之一。但我想知道是否有另一种方法来检查文件是否存在。

【问题讨论】:

  • +1 自己遇到了这个问题。我们最终在文件的公共地址上执行了 HTTP HEAD,但这不是一个通用的解决方案。

标签: python google-cloud-storage file-exists


【解决方案1】:

这篇文章已经过时了,您现在实际上可以使用 blob 类检查 GCP 上是否存在文件,但是因为我花了一段时间才找到答案,所以在这里添加给其他正在寻找解决方案的人

from google.cloud import storage

name = 'file_i_want_to_check.txt'   
storage_client = storage.Client()
bucket_name = 'my_bucket_name'
bucket = storage_client.bucket(bucket_name)
stats = storage.Blob(bucket=bucket, name=name).exists(storage_client)

文档是here

希望这会有所帮助!

编辑

根据@om-prakash 的评论,如果文件在文件夹中,则名称应包含文件的路径:

name = "folder/path_to/file_i_want_to_check.txt"

【讨论】:

  • 如果文件存在于谷歌云存储的某个文件夹中而不是云存储的根目录中,上述解决方案可能不起作用,请改为stats = storage.Blob(bucket=bucket, name="folder_1/another_folder_2/your_file.txt").exists(storage_client)
  • 谢谢!这正是我所需要的。
  • 如果 blob 是文件夹而不是文件,它也将不起作用。
【解决方案2】:

就像在 blob 对象中使用 exists 方法一样简单:

from google.cloud import storage

def blob_exists(projectname, credentials, bucket_name, filename):
   client = storage.Client(projectname, credentials=credentials)
   bucket = client.get_bucket(bucket_name)
   blob = bucket.blob(filename)
   return blob.exists()

【讨论】:

  • 对于数千个 URL,这很慢。是否可以一次性提交一批密钥/存储桶?
  • 如果文件很大(轶事),似乎也容易出错。 urllib3.exceptions.ProtocolError: ('Connection aborted.', OSError(0, 'Error'))
【解决方案3】:

@nickthefreak 提供的答案是正确的,Om Prakash 的评论也是如此。另一个注意事项是 bucket_name 不应在前面包含 gs:// 或在末尾包含 /

借鉴@nickthefreak 的示例和 Om Prakash 的评论:

from google.cloud import storage

name = 'folder1/another_folder/file_i_want_to_check.txt'   

storage_client = storage.Client()
bucket_name = 'my_bucket_name'  # Do not put 'gs://my_bucket_name'
bucket = storage_client.bucket(bucket_name)
stats = storage.Blob(bucket=bucket, name=name).exists(storage_client)

stats 将是一个布尔值(True 或 False),具体取决于文件是否存在于存储桶中。

(我没有足够的声望点来发表评论,但我想为其他人节省一些时间,因为我在这方面浪费了太多时间)。

【讨论】:

  • 这应该是公认的答案。 .exists() 不需要额外的参数。
【解决方案4】:

如果你正在寻找 NodeJS 中的解决方案,那么这里就是:

var storage = require('@google-cloud/storage')();
var myBucket = storage.bucket('my-bucket');

var file = myBucket.file('my-file');

file.exists(function(err, exists) {});

// If the callback is omitted, then this function return a Promise.
file.exists().then(function(data) {
  var exists = data[0];
});

如果您需要更多帮助,可以参考此文档: https://cloud.google.com/nodejs/docs/reference/storage/1.5.x/File#exists

【讨论】:

  • OP 专门要求 Python
【解决方案5】:

您可以使用 stat 函数来获取文件信息。实际上,这将向谷歌云存储发出 HEAD 请求,而不是 GET,这会占用较少的资源。

import cloudstorage as gcs
# return stat if there is one, else None or false. A stat record should be truthy
def is_file_available(filepath):

  try:
    return gcs.stat(filepath)
  except gcs_errors.NotFoundError as e:
    return False

【讨论】:

    【解决方案6】:

    如果您在“Google AI Platform”之类的服务上使用 gcs 文件,请使用 tensorflow 检查文件是否存在:

    import tensorflow as tf
    file_exists = tf.gfile.Exists('gs://your-bucket-name/your-file.txt')
    

    【讨论】:

      【解决方案7】:

      我猜没有函数可以直接检查给定路径的文件是否存在。
      我创建了一个函数,它使用files.listdir() API 函数列出存储桶中的所有文件并将其与我们想要的文件名匹配。如果找到则返回 true,否则返回 false。

      【讨论】:

        【解决方案8】:

        我在谷歌云存储上搜索的文件:init.sh

        完整路径: gs://cw-data/spark_app_code/init.sh

        >>> from google.cloud import storage
        
        >>> def is_exist(bucket_name,object):
        ...     client = storage.Client()
        ...     bucket = client.bucket(bucket_name)
        ...     blob = bucket.get_blob(object)
        ...     try:
        ...             return blob.exists(client)
        ...     except:
        ...             return False
        ...
        >>> is_exist('cw-data','spark_app_code')
            False
        >>> is_exist('cw-data','spark_app_code/')
            True
        >>> is_exist('cw-data','init.sh')
            False
        >>> is_exist('cw-data','spark_app_code/init.sh')
            True
        >>> is_exist('cw-data','/init.sh')
            False
        >>>
        

        在这里,文件不是以它们存储在本地文件系统上的方式存储的,而是作为键存储的。因此,在 google 存储上搜索文件时,请使用绝对路径,而不仅仅是文件名。

        【讨论】:

          【解决方案9】:

          您可以使用自定义函数(如下所示)检查文件是否存在

          def is_file_available(filepath):
           #check if the file is available
           fileavability = 'yes';
           try: 
            fp = files.open(filepath, 'r')
            fp.close()
           except Exception,e:
            fileavability = 'no'
           return fileavability 
          
          按以下方式使用上述功能
           filepath = '/gs/test/testme.txt'
           fileavability = is_file_available(filepath)
          

          注意:在上述函数中,当尝试读取文件的应用程序未授予读取权限时,您也可能得到“否”的结果。

          【讨论】:

            【解决方案10】:

            几年前 Amit 的回答略有不同,已针对 cloudstorage api 进行了更新。

            import cloudstorage as gcs
            
            def GCSExists(gcs_file):
                '''
                True if file exists; pass complete /bucket/file
                '''
                try:
                    file = gcs.open(gcs_file,'r')
                    file.close()
                    status = True
                except:
                    status = False
                return status
            

            【讨论】:

            • import cloudstorage as gcs gcs.open("gs://foo/foo.bar") AttributeError: module 'cloudstorage' has no attribute 'open'
            【解决方案11】:

            是的!有可能! from this

            这是我的代码:

            def get_by_signed_url(self, object_name, bucket_name=GCLOUD_BUCKET_NAME):
                bucket = self.client_storage.bucket(bucket_name)
                blob = bucket.blob(object_name)
            
                #this is check if file exist or not
                stats = blob.exists(self.client_storage)
                if not stats:
                    raise NotFound(messages.ERROR_NOT_FOUND)
            
                url_lifetime = self.expiration  # Seconds in an hour
                serving_url = blob.generate_signed_url(url_lifetime)
                return self.session.get(serving_url)
            

            【讨论】:

            • 虽然存储桶中有一个文件,但我总是返回“False”。
            • @AjitK'sagar 我不是。如果文件存在 gcs 将返回文件的 url。也许你的网址不正确?
            【解决方案12】:

            由于此问题上的 accepted answer 没有提供太多细节 - 这是一个使用 gsutil 的现代解决方案,其功能如该答案所述。

            如果您需要在脚本中多次查询 GCS 文件,这将比其他答案更有效。

            def bucket_to_list(bucketname: str):
                '''
                Return bucket's contents to python list of strings. 
                We also slice off the bucket name on each line, 
                in case we need to search many buckets for one file.
                '''
                return subprocess.run(['gsutil','ls','-r', bucketname + '**'], shell=False, text=True, stdout=subprocess.PIPE).stdout.replace(bucketname, "").splitlines()
            
            

            按以下方式使用:

            # call once for each bucket to store bucket contents 
            mybucket1 = 'gs://mybucket1/'
            mybucket1list = bucket_to_list(mybucket1)
            
            # limiting list to a bucket's "subdirectories"
            mybucket2 = 'gs://mybucket2/subdir1/subdir2/'
            mybucket2list = bucket_to_list(mybucket2)
            
            # example filename list to check, we dont need to add the gs:// paths 
            filestocheck = ['file1.ext', 'file2.ext', 'file3.ext']
            
            # check both buckets for files in our filelist
            for file in filestocheck:
                if file in mybucket1list:
                    # do something if file exists in bucket1
                elif file in mybucket2list:
                    # do something if file exists in bucket2
                else:
                    # do something if file doesn't exist in either bucket 
            

            【讨论】:

            • 我不确定这是“现代解决方案”。如果你使用的是 python,那么使用 GCP API 会更聪明。
            • 如果存在原生库,不要调用子进程
            【解决方案13】:

            从 google.cloud 导入存储

            def if_file_exists(name:str,bucket_name:str):
                storage_client = storage.Client()
                bucket = storage_client.bucket(bucket_name)
                stats = storage.Blob.from_string(f"gs://{bucket_name}/{name}").exists(storage_client)
                return stats
            

            打印(if_file_exists('audios/courses/ActivityPlaying/1320210506130438.wav',GC_BUCKET_NAME),">>>")

            name args 是文件的剩余路径

            if_file_exists 函数接受两个位置参数,第一个是对象键,第二个是存储桶名称,如果文件存在则返回 true,否则返回 false

            【讨论】:

              猜你喜欢
              • 2020-07-20
              • 1970-01-01
              • 2022-11-05
              • 1970-01-01
              • 2018-02-15
              • 2019-01-12
              • 1970-01-01
              • 2010-09-16
              • 2020-11-23
              相关资源
              最近更新 更多