本文整理汇总了Python中sinastorage.bucket.SCSBucket类的典型用法代码示例。如果您正苦于以下问题：Python SCSBucket类的具体用法？Python SCSBucket怎么用？Python SCSBucket使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了SCSBucket类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_put_file_relax
def test_put_file_relax(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
scs.put(self.object_key, 'fileContent...fileContent...fileContent...fileContent...fileContent...')
orign_file_metaResult = scs.meta(self.object_key)
canned_acl = 'public-read-write' #快捷ACL
metadata = {} #自定义文件属性信息
metadata['author'] = 'dage'
metadata['home'] = 'tianjin'
metadata['age'] = '18'
mimetype = 'text/plain'
scs.put_relax(key=self.object_key+'_relax', sina_sha1=orign_file_metaResult['Content-SHA1'],
s_sina_length=orign_file_metaResult['Size'], acl=canned_acl,
metadata=metadata, mimetype=mimetype)
relax_file_aclInfo = scs.acl_info(self.object_key+'_relax')
self.assertTrue(ACL.ACL_GROUP_ANONYMOUSE in relax_file_aclInfo['ACL'], 'The acl dose not contains GRPS000000ANONYMOUSE group')
self.assertTrue(ACL.ACL_READ in relax_file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t read right')
self.assertTrue(ACL.ACL_WRITE in relax_file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t write right')
relax_file_info = scs.info(self.object_key+'_relax')
#assert metadata
self.assertEqual(metadata['author'], relax_file_info['metadata']['author'], 'The response metadata[\'author\'] is not match')
self.assertEqual(metadata['home'], relax_file_info['metadata']['home'], 'The response metadata[\'home\'] is not match')
self.assertEqual(metadata['age'], relax_file_info['metadata']['age'], 'The response metadata[\'age\'] is not match')
#assert content-type
self.assertEqual(mimetype, relax_file_info['headers']['content-type'], 'The response content-type is not match')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:33，代码来源:test.py
示例2: run
def run(self):
try:
self.mutex.lock()
self.state = RunnableState.RUNNING
s = SCSBucket(self.bucketName)
if self.key.rfind('/') == len(self.key)-1 :
m = (("prefix", smart_str(self.key)),
("delimiter", '/'),
("max-keys", 5),
("formatter","json"))
args = dict((str(k), str(v)) for (k, v) in m if v is not None)
response = s.send(s.request(key='', args=args))
files_generator = SCSListing.parse(response)
if files_generator.contents_quantity > 0 or files_generator.common_prefixes_quantity > 0 :
for item in files_generator:
if cmp(item[0],self.key) != 0:
self.emitter.emit(QtCore.SIGNAL("DeleteObjectForbidden(PyQt_PyObject,PyQt_PyObject)"), self, u'不能删除非空目录(前缀) ')
return
# s = SCSBucket(self.bucketName)
self.response = s.send(s.request(method="DELETE", key=self.key))
except SCSError, e:
self.state = RunnableState.DID_FAILED
self.response = SCSResponse(e.urllib2Request, e.urllib2Response)
self.response._responseBody = e.data
self.emitter.emit(QtCore.SIGNAL("DeleteObjectDidFailed(PyQt_PyObject,PyQt_PyObject)"), self, e.msg)
return
开发者ID:vfhky，项目名称:SCS_Client，代码行数:28，代码来源:Runnables.py
示例3: copy_file
def copy_file():
s = SCSBucket('asdasdasdasd')
"""
注意：
source 必须从bucket开始，如：'/cloud0/aaa.txt'
"""
s.copy('/asdasdasdasd/aaaa.txt', 'aaaa22111.txt')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:7，代码来源:sample.py
示例4: update_file_acl
def update_file_acl():
s = SCSBucket('test11')
acl = {}
acl[ACL.ACL_GROUP_ANONYMOUSE] = [ACL.ACL_READ]
acl[ACL.ACL_GROUP_CANONICAL] = [ACL.ACL_READ_ACP,ACL.ACL_WRITE_ACP]
s.update_acl('sss.txt', acl)
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:7，代码来源:sample.py
示例5: test_remove_bucket
def test_remove_bucket(self):
scs = SCSBucket(self.bucket_name)
scs.put_bucket()
try:
scs.delete_bucket()
except:
self.fail('Remove bucket:%s is failed'%self.bucket_name)
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:8，代码来源:test.py
示例6: test_list_bucket
def test_list_bucket(self):
scs = SCSBucket()
bucket_gen = scs.list_buckets()
self.assertTrue(bucket_gen is not None, 'List bucket result is None')
for bucket in bucket_gen:
bucketStr = str(bucket)
self.assertTrue(bucket[0] is not None, 'bucket:%s name is None'%bucketStr)
self.assertTrue(bucket[1] is not None, 'bucket:%s CreationDate is None'%bucketStr)
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:8，代码来源:test.py
示例7: listFile
def listFile(bucketName):
Files = []
s = SCSBucket(bucketName)
files = s.listdir()
for i in files:
#i = (name, isPrefix, sha1, expiration_time, modify, owner, md5, content_type, size)
Files.append(i)
return Files
开发者ID:gothack329，项目名称:sirius.py，代码行数:8，代码来源:scs_sync.py
示例8: listBucket
def listBucket():
Buckets = []
s = SCSBucket()
buckets_generator = s.list_buckets()
for bucket in buckets_generator:
bucketName = bucket[0]
bMeta = SCSBucket(bucketName).meta()
Buckets.append(bMeta)
return Buckets
开发者ID:gothack329，项目名称:sirius.py，代码行数:9，代码来源:scs_sync.py
示例9: test_copy_file
def test_copy_file(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
scs.putFile(self.object_key, self.local_file_name)
canned_acl = 'public-read' #快捷ACL
metadata = {} #自定义文件属性信息
metadata['author'] = 'copied'
metadata['home'] = 'beijing'
metadata['age'] = '189'
mimetype = 'text/plain'
scs.copy(source='/'+self.bucket_name+'/'+self.object_key,
key=self.object_key+'_copied', acl=canned_acl,
metadata=metadata, mimetype=mimetype)
copied_file_aclInfo = scs.acl_info(self.object_key+'_copied')
self.assertTrue(ACL.ACL_GROUP_ANONYMOUSE in copied_file_aclInfo['ACL'], 'The acl dose not contains GRPS000000ANONYMOUSE group')
self.assertTrue(ACL.ACL_READ in copied_file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t read right')
copied_file_info = scs.info(self.object_key+'_copied')
#assert metadata
self.assertEqual(metadata['author'], copied_file_info['metadata']['author'], 'The response metadata[\'author\'] is not match')
self.assertEqual(metadata['home'], copied_file_info['metadata']['home'], 'The response metadata[\'home\'] is not match')
self.assertEqual(metadata['age'], copied_file_info['metadata']['age'], 'The response metadata[\'age\'] is not match')
#assert content-type
self.assertEqual(mimetype, copied_file_info['headers']['content-type'], 'The response content-type is not match')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:30，代码来源:test.py
示例10: test_update_file_meta
def test_update_file_meta(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
content = u'this is a file content text!!'
scs.put(self.object_key, content)
canned_acl = 'public-read-write' #快捷ACL
metadata = {} #自定义文件属性信息
metadata['author'] = 'dage'
metadata['home'] = 'tianjin'
metadata['age'] = '18'
mimetype = 'text/plain'
scs.update_meta(self.object_key, metadata=metadata, acl=canned_acl, mimetype=mimetype)
file_info = scs.info(self.object_key)
#assert metadata
self.assertEqual(metadata['author'], file_info['metadata']['author'], 'The response metadata[\'author\'] is not match')
self.assertEqual(metadata['home'], file_info['metadata']['home'], 'The response metadata[\'home\'] is not match')
self.assertEqual(metadata['age'], file_info['metadata']['age'], 'The response metadata[\'age\'] is not match')
#assert content-type
# print file_info['headers']
#TODO:服务器问题 self.assertEqual(mimetype, file_info['headers']['content-type'], 'The response content-type is not match')
#acl
file_aclInfo = scs.acl_info(self.object_key)
self.assertTrue(ACL.ACL_GROUP_ANONYMOUSE in file_aclInfo['ACL'], 'The acl dose not contains GRPS000000ANONYMOUSE group')
self.assertTrue(ACL.ACL_READ in file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t read right')
self.assertTrue(ACL.ACL_WRITE in file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t write right')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:30，代码来源:test.py
示例11: test_create_bucket_with_acl
def test_create_bucket_with_acl(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
metaResult = scs.meta()
self.assertTrue(ACL.ACL_GROUP_ANONYMOUSE in metaResult['ACL'], 'The bucket:%s acl dose not contains GRPS000000ANONYMOUSE group'%self.bucket_name)
self.assertTrue(ACL.ACL_READ in metaResult['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The bucket:%s acl GRPS000000ANONYMOUSE group hasn\'t read right'%self.bucket_name)
self.assertTrue(ACL.ACL_WRITE in metaResult['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The bucket:%s acl GRPS000000ANONYMOUSE group hasn\'t write right'%self.bucket_name)
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:11，代码来源:test.py
示例12: upload
def upload(f,bucketName,localDir,ss=0,fl=0):
f = f.strip('/')
s = SCSBucket(bucketName)
failed = None
try:
result = s.putFile(f,localDir+f)
print f+' upload success.'
ss += 1
except Exception,e:
fl += 1
failed = f
开发者ID:gothack329，项目名称:sirius.py，代码行数:11，代码来源:scs_sync.py
示例13: obj_upload
def obj_upload(data, fileTag):
""" Upload photo by object. """
sinastorage.setDefaultAppInfo(API_KEY, API_SECRET)
s = SCSBucket('sharephotos')
path = time.ctime().replace(' ', '_')
# if '/' in file name there will be problems
filename = fileTag[:6] + '_' + path + '.jpg' # use jpg temporary
scs_response = s.put(filename, data)
s.update_acl(filename, acl)
url = s.make_url(filename) # get url of image in the storage
return url
开发者ID:RawEvan，项目名称:sharephotos，代码行数:13，代码来源:storage.py
示例14: test_update_file_acl
def test_update_file_acl(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
content = u'this is a file content text!!'
canned_acl = 'private' #快捷ACL
scs.put(key=self.object_key, data=content, acl=canned_acl)
#设置ACL
acl = {}
acl[ACL.ACL_GROUP_ANONYMOUSE] = [ACL.ACL_READ, ACL.ACL_WRITE, ACL.ACL_READ_ACP]
acl[ACL.ACL_GROUP_CANONICAL] = [ACL.ACL_READ_ACP,ACL.ACL_WRITE_ACP, ACL.ACL_READ]
user_id = 'SINA0000001001AABBCC'
acl[user_id] = [ACL.ACL_WRITE, ACL.ACL_READ]
scs.update_acl(self.object_key, acl)
#assert
file_aclInfo = scs.acl_info(self.object_key)
#ACL_GROUP_ANONYMOUSE
self.assertTrue(ACL.ACL_GROUP_ANONYMOUSE in file_aclInfo['ACL'], 'The acl dose not contains GRPS000000ANONYMOUSE group')
self.assertTrue(ACL.ACL_READ in file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t read right')
self.assertTrue(ACL.ACL_WRITE in file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t write right')
self.assertTrue(ACL.ACL_READ_ACP in file_aclInfo['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The acl GRPS000000ANONYMOUSE group hasn\'t read_acp right')
#ACL_GROUP_CANONICAL
self.assertTrue(ACL.ACL_GROUP_CANONICAL in file_aclInfo['ACL'], 'The acl dose not contains GRPS0000000CANONICAL group')
self.assertTrue(ACL.ACL_READ in file_aclInfo['ACL'][ACL.ACL_GROUP_CANONICAL], 'The acl GRPS0000000CANONICAL group hasn\'t read right')
self.assertTrue(ACL.ACL_WRITE_ACP in file_aclInfo['ACL'][ACL.ACL_GROUP_CANONICAL], 'The acl GRPS0000000CANONICAL group hasn\'t write_acp right')
self.assertTrue(ACL.ACL_READ_ACP in file_aclInfo['ACL'][ACL.ACL_GROUP_CANONICAL], 'The acl GRPS0000000CANONICAL group hasn\'t read_acp right')
#user_id
self.assertTrue(user_id in file_aclInfo['ACL'], 'The acl dose not contains user_id group')
self.assertTrue(ACL.ACL_READ in file_aclInfo['ACL'][user_id], 'The acl user_id group hasn\'t read right')
self.assertTrue(ACL.ACL_WRITE in file_aclInfo['ACL'][user_id], 'The acl user_id group hasn\'t write right')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:33，代码来源:test.py
示例15: url_upload
def url_upload(imgSrc='http://www.w3school.com.cn/i/site_photoref.jpg'):
""" Upload photo by url. """
sinastorage.setDefaultAppInfo(API_KEY, API_SECRET)
s = SCSBucket('sharephotos')
data = urllib2.urlopen(imgSrc).read()
path = imgSrc.split('/')[2] + '/%s__' % time.ctime()
# if '/' in file name there will be problems
filename = path + imgSrc.replace('/', '_')
scs_response = s.put(filename, data)
s.update_acl(filename, acl)
url = s.make_url(filename) # get url of image in the storage
return url
开发者ID:RawEvan，项目名称:sharephotos，代码行数:14，代码来源:storage.py
示例16: test_get_bucket_meta
def test_get_bucket_meta(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
metaResult = scs.meta()
self.assertTrue(ACL.ACL_GROUP_ANONYMOUSE in metaResult['ACL'], 'The bucket:%s acl dose not contains GRPS000000ANONYMOUSE group'%self.bucket_name)
self.assertTrue(ACL.ACL_READ in metaResult['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The bucket:%s acl GRPS000000ANONYMOUSE group hasn\'t read right'%self.bucket_name)
self.assertTrue(ACL.ACL_WRITE in metaResult['ACL'][ACL.ACL_GROUP_ANONYMOUSE], 'The bucket:%s acl GRPS000000ANONYMOUSE group hasn\'t write right'%self.bucket_name)
self.assertEqual(self.bucket_name, metaResult['Project'], 'The metaResult[\'Project\'] is not equal %s'%self.bucket_name)
self.assertTrue(metaResult['Owner'] is not None, 'The metaResult[\'Owner\'] is None')
self.assertTrue(metaResult['Last-Modified'] is not None, 'The metaResult[\'Last-Modified\'] is None')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:15，代码来源:test.py
示例17: list_bucket_files
def list_bucket_files():
s = SCSBucket('test11')
files_generator = s.listdir()#delimiter='/')
print '-----list_bucket_files---------'
print '-----detail---------'
print ('truncated : %r\n'
'marker:%r\n'
'prefix:%r\n'
'delimiter:%r\n'
'contents_quantity:%r\n'
'common_prefixes_quantity:%r\n'
'next_marker:%r'%(files_generator.truncated,
files_generator.marker,
files_generator.prefix,
files_generator.delimiter,
files_generator.contents_quantity,
files_generator.common_prefixes_quantity,
files_generator.next_marker))
print '-----file list---------'
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:19，代码来源:sample.py
示例18: upload
def upload(self):
''' 普通上传 '''
try:
self.mutex.lock()
self.state = RunnableState.RUNNING
s = SCSBucket(self.bucketName)
self.fileWithCallback = FileWithCallback(self.filePath, 'rb', self.uploadCallBack)
scsResponse = s.putFileByHeaders('%s%s'%(self.prefix,os.path.basename(self.filePath)), self.fileWithCallback)
self.response = scsResponse
except SCSError, e:
self.response = SCSResponse(e.urllib2Request, e.urllib2Response)
if isinstance(e, ManualCancel): #手动取消
self.state = RunnableState.DID_CANCELED
self.response._responseBody = u'手动取消'
self.emitter.emit(QtCore.SIGNAL("fileUploadDidCanceled(PyQt_PyObject,PyQt_PyObject)"), self, e.msg)
else:
self.state = RunnableState.DID_FAILED
self.response._responseBody = e.data
self.emitter.emit(QtCore.SIGNAL("fileUploadDidFailed(PyQt_PyObject,PyQt_PyObject)"), self, e.msg)
return
开发者ID:vfhky，项目名称:SCS_Client，代码行数:21，代码来源:Runnables.py
示例19: test_get_file_meta
def test_get_file_meta(self):
scs = SCSBucket(self.bucket_name)
try:
scs.put_bucket('public-read-write')
except:
self.fail('create bucket:%s is failed'%self.bucket_name)
metadata = {} #自定义文件属性信息
metadata['author'] = 'dage'
metadata['home'] = 'tianjin'
metadata['age'] = '18'
mimetype = 'text/plain'
scs.putFile(key=self.object_key, filePath=self.local_file_name,
metadata=metadata, mimetype=mimetype)
metaResult = scs.meta(self.object_key)
self.assertEqual(self.object_key, metaResult['File-Name'], 'The meta[\'File-Name\'] is not equals '+self.object_key)
self.assertEqual(mimetype, metaResult['Type'], 'The meta[\'Type\'] is not equals '+mimetype)
self.assertTrue(metaResult['Content-SHA1'] is not None, 'The meta[\'Content-SHA1\'] is None')
self.assertTrue(metaResult['Content-MD5'] is not None, 'The meta[\'Content-MD5\'] is None')
self.assertTrue(metaResult['Owner'] is not None, 'The meta[\'Owner\'] is None')
self.assertTrue('x-amz-meta-author' in metaResult['File-Meta'], 'File-Meta dose not contains x-amz-meta-author key')
self.assertEqual(metadata['author'], metaResult['File-Meta']['x-amz-meta-author'], 'The metaResult[\'File-Meta\'][\'x-amz-meta-author\'] value is not match')
self.assertTrue('x-amz-meta-home' in metaResult['File-Meta'], 'File-Meta dose not contains x-amz-meta-home key')
self.assertEqual(metadata['home'], metaResult['File-Meta']['x-amz-meta-home'], 'The metaResult[\'File-Meta\'][\'x-amz-meta-home\'] value is not match')
self.assertTrue('x-amz-meta-age' in metaResult['File-Meta'], 'File-Meta dose not contains x-amz-meta-age key')
self.assertEqual(metadata['age'], metaResult['File-Meta']['x-amz-meta-age'], 'The metaResult[\'File-Meta\'][\'x-amz-meta-age\'] value is not match')
开发者ID:SinaCloudStorage，项目名称:SinaStorage-SDK-Python，代码行数:30，代码来源:test.py
示例20: storeImage
def storeImage(imgSrc):
sinastorage.setDefaultAppInfo("1cjfyo5kQPdnsI3cUc6W", "a3c139370a3509f269331930515729747573aa10")
djBucket = SCSBucket("dj-images") # not dj_images
acl = {} # Access control list
acl[ACL.ACL_GROUP_ANONYMOUSE] = [ACL.ACL_READ]
data = urllib2.urlopen(imgSrc).read()
path = imgSrc.split("/")[2] + "/"
# if '/' in file name there will be problems
filename = path + imgSrc.replace("/", "@")
scsResponse = djBucket.put(filename, data) # upload the file
stUrl = djBucket.make_url(filename) # get url of image in the storage
djBucket.update_acl(filename, acl) # set acl for the file
# save infomation to sql
try:
imgstorage.objects.get_or_create(original_url=imgSrc, storage_url=stUrl)
except:
print "insert into mysql error"
pass # solve this later
return stUrl
开发者ID:AugustLONG，项目名称:images_spider，代码行数:22，代码来源:storage.py
注：本文中的sinastorage.bucket.SCSBucket类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论