我们有一项工作,检查云存储中的文件是否已被修改。如果是,则读取文件中的数据并进一步处理。
我想知道是否有API可以检查云存储中的文件上次修改时间。
我想知道是否有API可以检查云存储中的文件上次修改时间。
from google.cloud import storage
def blob_metadata(bucket_name, blob_name):
"""Prints out a blob's metadata."""
# bucket_name = 'your-bucket-name'
# blob_name = 'your-object-name'
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.get_blob(blob_name)
print("Blob: {}".format(blob.name))
print("Bucket: {}".format(blob.bucket.name))
print("Storage class: {}".format(blob.storage_class))
print("ID: {}".format(blob.id))
print("Size: {} bytes".format(blob.size))
print("Updated: {}".format(blob.updated))
print("Generation: {}".format(blob.generation))
print("Metageneration: {}".format(blob.metageneration))
print("Etag: {}".format(blob.etag))
print("Owner: {}".format(blob.owner))
print("Component count: {}".format(blob.component_count))
print("Crc32c: {}".format(blob.crc32c))
print("md5_hash: {}".format(blob.md5_hash))
print("Cache-control: {}".format(blob.cache_control))
print("Content-type: {}".format(blob.content_type))
print("Content-disposition: {}".format(blob.content_disposition))
print("Content-encoding: {}".format(blob.content_encoding))
print("Content-language: {}".format(blob.content_language))
print("Metadata: {}".format(blob.metadata))
print("Temporary hold: ", "enabled" if blob.temporary_hold else "disabled")
print(
"Event based hold: ",
"enabled" if blob.event_based_hold else "disabled",
)
if blob.retention_expiration_time:
print(
"retentionExpirationTime: {}".format(
blob.retention_expiration_time
)
)
blob.updated
属性。你可以使用boto来完成此操作:
>>> import boto
>>> conn = boto.connect_gs()
>>> bucket = conn.get_bucket('yourbucket')
>>> k = bucket.get_key('yourkey')
>>> k.last_modified
'Tue, 04 Dec 2012 17:44:57 GMT'
还有一个App Engine Python接口可以访问云存储,但我认为它没有暴露你想要的元数据。
云存储有一个API,您可以使用它来获取对象的创建时间
请参见https://developers.google.com/storage/docs/json_api/v1/objects
我正在使用 @orby 上面提到的解决方案,使用 blob.updated
获取最新文件。 但是存储桶中有450多个文件,而此脚本需要大约6-7分钟才能遍历所有文件并提供最新文件。 我想blob.updated
部分需要一些时间来处理。 有更快的方法吗?
files = bucket.list_blobs()
fileList = [file.name for file in files if '.dat' in file.name]
latestFile = fileList[0]
latestTimeStamp = bucket.get_blob(fileList[0]).updated
for i in range(len(fileList)):
timeStamp = bucket.get_blob(fileList[i]).updated
if timeStamp > latestTimeStamp:
latestFile = fileList[i]
latestTimeStamp = timeStamp
print(latestFile)
bucket.get_blob(fileList[i]).updated
,因为您已经在 fileList 中拥有了该 Blob。然后,您可以调用 i.updated
。 - orbyfiles = bucket.list_blobs()
fileList = [file for file in files if '.dat' in file.name]
latestFile = None
for i in fileList:
if not latestFile:
latestFile = i
continue
if i.updated > latestFile.updated:
latestFile = i
print(latestFile)
- orby