Shulou(Shulou.com)06/01 Report--

Editor to share with you how to achieve python query bucket has been used script, I believe that most people do not know much about it, so share this article for your reference, I hope you will learn a lot after reading this article, let's go to understand it!

Currently, only ceph's S3 scheme is supported. For specific configuration, please see the instructions.

#-*-coding: utf-8-*-import requestsimport jsonfrom email.utils import formatdateimport hmacpy3k = Falsefrom hashlib import sha1 as shatry: from urlparse import urlparse from base64 import encodestringexcept: py3k = True from urllib.parse import urlparse from base64 import encodebytes as encodestringclass AuthBase (object): "" Base class that all auth implementations derive from "" def _ _ call__ (self) R): raise NotImplementedError ('Auth hooks must be callable.') class S3Auth (AuthBase): "" Attaches AWS Authentication to the given Request object. "" Service_base_url = 's3.amazonaws.com' # List of Query String Arguments of Interest special_params = ['acl',' location', 'logging',' partNumber', 'policy',' requestPayment', 'torrent',' versioning', 'versionId',' website', 'uploads',' uploadId', 'response-content-type',' response-content-language', 'response-expires' 'response-cache-control',' delete', 'lifecycle',' response-content-disposition', 'response-content-encoding'] def _ _ init__ (self, access_key, secret_key Service_url=None): if service_url: self.service_base_url = service_url self.access_key = str (access_key) self.secret_key = str (secret_key) self.au = "" def _ call__ (self, r): # Create date header if it is not created yet. If not 'date' in r.headers and not' x Muhammad date'in r.headers: r.headers ['date'] = formatdate (timeval=None, localtime=False) Usegmt=True) signature = self.get_signature (r) if py3k: signature = signature.decode ('utf-8') r.headers [' Authorization'] = 'AWS% SV% s'% (self.access_key, signature) self.au = r.headers # print self.au return r def get_signature (self) R): canonical_string = self.get_canonical_string (r.url, r.headers, r.method) if py3k: key = self.secret_key.encode ('utf-8') msg = canonical_string.encode (' utf-8') else: key = self.secret_key msg = canonical_string h = hmac.new (key, msg) Digestmod=sha) return encodestring (h.digest ()). Strip () def get_canonical_string (self, url, headers) Method): parsedurl = urlparse (url) objectkey = parsedurl.path [1:] query_args = sorted (parsedurl.query.split ('&')) bucket = parsedurl.netloc [:-len (self.service_base_url)] if len (bucket) > 1: # remove last dot bucket = bucket [:-1] interesting_headers = {'content-md5':'' 'content-type':'' 'date':'} for key in headers: lk = key.lower () try: lk = lk.decode ('utf-8') except: pass if headers [key] and (lk in interesting_headers.keys () or lk.startswith (' Xmuramzmuri')): Interesting_ headers [LK] = headers [key] .strip () # If x-amz-date is used it supersedes the date header. If not py3k: in interesting_headers: interesting_headers ['date'] =' 'else: if' x in interesting_headers: interesting_headers ['date'] ='% s\ n'% method for key in sorted (interesting_headers.keys ()): Val = interesting_ headers [key] if key.startswith ('xMuamamzmuri'): buf + ='% s v% s\ n'% (key Val) else: buf + ='% s\ n'% val # append the bucket if it exists if bucket! ='': buf + ='/% s'% bucket # add the objectkey. Even if it doesn't exist Add the slash buf + ='/% s'% objectkey params_found = False # handle special query string arguments for q in query_args: K = q.split ('=') [0] if k in self.special_params: if params_found: buf + ='&% s'% Q else: Buf + ='?% s'% Q params_found = True return bufclass S3Admin (): def _ _ init__ (self): self.access_key ='# fill in access_key self.secret_key ='# fill in secret_key self.endpoint = 's3.ceph.work' # fill in endpoint def get_bucket_usage (self Bucketname): # url = 'http://%s/%s/'% (self.endpoint, bucketname) # path style url =' http://%s.%s/'% (bucketname,self.endpoint) # virtual hosted styple r = requests.head (url, auth=S3Auth (self.access_key, self.secret_key) Self.endpoint) print r.headers return r.headerss3client = S3Admin () bucket_name= 'xxx' # replace with the corresponding bucket name result = s3client.get_bucket_usage (bucket_name) print' objects_num=% s, total_Bytes_Used=% s'% (result ['Xmuri RGW' ObjectMerle country'], result ['XMuk RGWMI BytesUsed']) # Note that objects_num is the number of object in the current bucket Total_Bytes_Used is the used capacity in the current bucket (in Byte). These are all the contents of the article "how to query the bucket used script by python". Thank you for reading! I believe we all have a certain understanding, hope to share the content to help you, if you want to learn more knowledge, welcome to follow the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.