python编写api调用ceph对象网关

#_*_coding:utf-8_*_
import boto
import boto.s3.connection
#pip install filechunkio
from filechunkio import  FileChunkIO
import math
import  threading
import os
import Queue

class Chunk(object):
    num = 0
    offset = 0
    len = 0
    def __init__(self,n,o,l):
        self.num=n
        self.offset=o
        self.length=l

class CONNECTION(object):
    def __init__(self,access_key,secret_key,ip,port,is_secure=False,chrunksize=4<<20):
        self.conn=boto.connect_s3(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        host=ip,port=port,
        is_secure=is_secure,
        calling_format=boto.s3.connection.OrdinaryCallingFormat()
        )
        self.chrunksize=chrunksize

    #查询
    def list_all(self):
        all_buckets=self.conn.get_all_buckets()
        for bucket in all_buckets:
            print u‘容器名: %s‘ %(bucket.name)
            for key in bucket.list():
                print ‘ ‘*5,"%-20s%-20s%-20s%-40s%-20s" %(key.mode,key.owner.id,key.size,key.last_modified.split(‘.‘)[0],key.name)

    def list_single(self,bucket_name):
        try:
            single_bucket = self.conn.get_bucket(bucket_name)
        except Exception as e:
            print ‘bucket %s is not exist‘ %bucket_name
            return
        print u‘容器名: %s‘ % (single_bucket.name)
        for key in single_bucket.list():
            print ‘ ‘ * 5, "%-20s%-20s%-20s%-40s%-20s" % (key.mode, key.owner.id, key.size, key.last_modified.split(‘.‘)[0], key.name)

    #普通小文件下载：文件大小<=4M
    def dowload_file(self,filepath,key_name,bucket_name):
        all_bucket_name_list = [i.name for i in self.conn.get_all_buckets()]
        if bucket_name not in all_bucket_name_list:
            print ‘Bucket %s is not exist,please try again‘ % (bucket_name)
            return
        else:
            bucket = self.conn.get_bucket(bucket_name)

        all_key_name_list = [i.name for i in bucket.get_all_keys()]
        if key_name not in all_key_name_list:
            print ‘File %s is not exist,please try again‘ % (key_name)
            return
        else:
            key = bucket.get_key(key_name)

        if not os.path.exists(os.path.dirname(filepath)):
            print ‘Filepath %s is not exists, sure to create and try again‘ % (filepath)
            return

        if os.path.exists(filepath):
            while True:
                d_tag = raw_input(‘File %s already exists, sure you want to cover (Y/N)?‘ % (key_name)).strip()
                if d_tag not in [‘Y‘, ‘N‘] or len(d_tag) == 0:
                    continue
                elif d_tag == ‘Y‘:
                    os.remove(filepath)
                    break
                elif d_tag == ‘N‘:
                    return
        os.mknod(filepath)
        try:
            key.get_contents_to_filename(filepath)
        except Exception:
            pass

    # 普通小文件上传：文件大小<=4M
    def upload_file(self,filepath,key_name,bucket_name):
        try:
            bucket = self.conn.get_bucket(bucket_name)
        except Exception as e:
            print ‘bucket %s is not exist‘ % bucket_name
            tag = raw_input(‘Do you want to create the bucket %s: (Y/N)?‘ % bucket_name).strip()
            while tag not in [‘Y‘, ‘N‘]:
                tag = raw_input(‘Please input (Y/N)‘).strip()
            if tag == ‘N‘:
                return
            elif tag == ‘Y‘:
                self.conn.create_bucket(bucket_name)
                bucket = self.conn.get_bucket(bucket_name)
        all_key_name_list = [i.name for i in bucket.get_all_keys()]
        if key_name in all_key_name_list:
            while True:
                f_tag = raw_input(u‘File already exists, sure you want to cover (Y/N)?: ‘).strip()
                if f_tag not in [‘Y‘, ‘N‘] or len(f_tag) == 0:
                    continue
                elif f_tag == ‘Y‘:
                    break
                elif f_tag == ‘N‘:
                    return
        key=bucket.new_key(key_name)
        if not os.path.exists(filepath):
            print ‘File %s does not exist, please make sure you want to upload file path and try again‘ %(key_name)
            return
        try:
            f=file(filepath,‘rb‘)
            data=f.read()
            key.set_contents_from_string(data)
        except Exception:
            pass

    def delete_file(self,key_name,bucket_name):
        all_bucket_name_list = [i.name for i in self.conn.get_all_buckets()]
        if bucket_name not in all_bucket_name_list:
            print ‘Bucket %s is not exist,please try again‘ % (bucket_name)
            return
        else:
            bucket = self.conn.get_bucket(bucket_name)

        all_key_name_list = [i.name for i in bucket.get_all_keys()]
        if key_name not in all_key_name_list:
            print ‘File %s is not exist,please try again‘ % (key_name)
            return
        else:
            key = bucket.get_key(key_name)

        try:
            bucket.delete_key(key.name)
        except Exception:
            pass

    def delete_bucket(self,bucket_name):
        all_bucket_name_list = [i.name for i in self.conn.get_all_buckets()]
        if bucket_name not in all_bucket_name_list:
            print ‘Bucket %s is not exist,please try again‘ % (bucket_name)
            return
        else:
            bucket = self.conn.get_bucket(bucket_name)

        try:
            self.conn.delete_bucket(bucket.name)
        except Exception:
            pass

    #队列生成
    def init_queue(self,filesize,chunksize):   #8<<20 :8*2**20
        chunkcnt=int(math.ceil(filesize*1.0/chunksize))
        q=Queue.Queue(maxsize=chunkcnt)
        for i in range(0,chunkcnt):
            offset=chunksize*i
            length=min(chunksize,filesize-offset)
            c=Chunk(i+1,offset,length)
            q.put(c)
        return q

    #分片上传object
    def upload_trunk(self,filepath,mp,q,id):
        while not q.empty():
            chunk=q.get()
            fp=FileChunkIO(filepath,‘r‘,offset=chunk.offset,bytes=chunk.length)
            mp.upload_part_from_file(fp,part_num=chunk.num)
            fp.close()
            q.task_done()

    #文件大小获取---->S3分片上传对象生成----->初始队列生成(--------------->文件切，生成切分对象)
    def upload_file_multipart(self,filepath,key_name,bucket_name,threadcnt=8):
        filesize=os.stat(filepath).st_size
        try:
            bucket=self.conn.get_bucket(bucket_name)
        except Exception as e:
            print ‘bucket %s is not exist‘ % bucket_name
            tag=raw_input(‘Do you want to create the bucket %s: (Y/N)?‘ %bucket_name).strip()
            while tag not in [‘Y‘,‘N‘]:
                tag=raw_input(‘Please input (Y/N)‘).strip()
            if tag == ‘N‘:
                return
            elif tag == ‘Y‘:
                self.conn.create_bucket(bucket_name)
                bucket = self.conn.get_bucket(bucket_name)
        all_key_name_list=[i.name for i in bucket.get_all_keys()]
        if key_name  in all_key_name_list:
            while True:
                f_tag=raw_input(u‘File already exists, sure you want to cover (Y/N)?: ‘).strip()
                if f_tag not in [‘Y‘,‘N‘] or len(f_tag) == 0:
                    continue
                elif f_tag == ‘Y‘:
                    break
                elif f_tag == ‘N‘:
                    return

        mp=bucket.initiate_multipart_upload(key_name)
        q=self.init_queue(filesize,self.chrunksize)
        for i in range(0,threadcnt):
            t=threading.Thread(target=self.upload_trunk,args=(filepath,mp,q,i))
            t.setDaemon(True)
            t.start()
        q.join()
        mp.complete_upload()

    #文件分片下载
    def download_chrunk(self,filepath,key_name,bucket_name,q,id):
        while not q.empty():
            chrunk=q.get()
            offset=chrunk.offset
            length=chrunk.length
            bucket=self.conn.get_bucket(bucket_name)
            resp=bucket.connection.make_request(‘GET‘,bucket_name,key_name,headers={‘Range‘:"bytes=%d-%d" %(offset,offset+length)})
            data=resp.read(length)
            fp=FileChunkIO(filepath,‘r+‘,offset=chrunk.offset,bytes=chrunk.length)
            fp.write(data)
            fp.close()
            q.task_done()

    def download_file_multipart(self,filepath,key_name,bucket_name,threadcnt=8):
        all_bucket_name_list=[i.name for i in self.conn.get_all_buckets()]
        if bucket_name not in all_bucket_name_list:
            print ‘Bucket %s is not exist,please try again‘ %(bucket_name)
            return
        else:
            bucket=self.conn.get_bucket(bucket_name)

        all_key_name_list = [i.name for i in bucket.get_all_keys()]
        if key_name not in all_key_name_list:
            print ‘File %s is not exist,please try again‘ %(key_name)
            return
        else:
            key=bucket.get_key(key_name)

        if not os.path.exists(os.path.dirname(filepath)):
            print ‘Filepath %s is not exists, sure to create and try again‘ % (filepath)
            return

        if os.path.exists(filepath):
            while True:
                d_tag = raw_input(‘File %s already exists, sure you want to cover (Y/N)?‘ % (key_name)).strip()
                if d_tag not in [‘Y‘, ‘N‘] or len(d_tag) == 0:
                    continue
                elif d_tag == ‘Y‘:
                    os.remove(filepath)
                    break
                elif d_tag == ‘N‘:
                    return
        os.mknod(filepath)
        filesize=key.size
        q=self.init_queue(filesize,self.chrunksize)
        for i in range(0,threadcnt):
            t=threading.Thread(target=self.download_chrunk,args=(filepath,key_name,bucket_name,q,i))
            t.setDaemon(True)
            t.start()
        q.join()

if __name__ == ‘__main__‘:
    #约定：
    #1:filepath指本地文件的路径(上传路径or下载路径),指的是绝对路径
    #2:bucket_name相当于文件在对象存储中的目录名或者索引名
    #3:key_name相当于文件在对象存储中对应的文件名或文件索引

    access_key = "65IY4EC1BSFYNH6SHWGW"
    secret_key = "viNfIftLHhrPt2MYK44DkWGvxZb82aYqLrCzGYLx"
    ip=‘172.16.201.36‘
    port=8080
    conn=CONNECTION(access_key,secret_key,ip,port)
    #查看所有bucket以及其包含的文件
    #conn.list_all()

    #简单上传,用于文件大小<=4M
    # conn.upload_file(‘/etc/passwd‘,‘passwd‘,‘test_bucket01‘)
    #查看单一bucket下所包含的文件信息
    # conn.list_single(‘test_bucket01‘)

    #简单下载,用于文件大小<=4M
    # conn.dowload_file(‘/lhf_test/test01‘,‘passwd‘,‘test_bucket01‘)
    # conn.list_single(‘test_bucket01‘)

    #删除文件
    # conn.delete_file(‘passwd‘,‘test_bucket01‘)
    # conn.list_single(‘test_bucket01‘)
    #
    #删除bucket
    # conn.delete_bucket(‘test_bucket01‘)
    # conn.list_all()

    #切片上传(多线程),用于文件大小>4M,4M可修改
    # conn.upload_file_multipart(‘/etc/passwd‘,‘passwd_multi_upload‘,‘test_bucket01‘)
    # conn.list_single(‘test_bucket01‘)

    # 切片下载(多线程),用于文件大小>4M,4M可修改
    # conn.download_file_multipart(‘/lhf_test/passwd_multi_dowload‘,‘passwd_multi_upload‘,‘test_bucket01‘)