python - 如何检查文件是否存在于 Google Cloud Storage 中？

Question

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

我尝试使用os.path.exists(file_path)where file_path = "/gs/testbucket"，但出现文件未找到错误。

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

Answer 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)

文档在这里

希望这可以帮助！

编辑

根据@om-prakash 的评论，如果文件位于文件夹中，则名称应包含文件的路径：

name = "folder/path_to/file_i_want_to_check.txt"

Answer 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()

Answer 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 将是一个布尔值（真或假），具体取决于文件是否存在于存储桶中。

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

Answer 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

Answer 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

Answer 6

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

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

Answer 7

我在谷歌云存储上搜索的文件：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
>>>

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

Answer 8

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

Answer 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)

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

Answer 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

Answer 11

是的！有可能！由此

这是我的代码：

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)

Answer 12

由于这个问题的公认答案没有提供太多细节 - 这是一个使用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 

Answer 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

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

name args 是文件的剩余路径

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