查询bucket已用量脚本-python
时间:2022-04-25
本文章向大家介绍查询bucket已用量脚本-python,主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
目前仅支持ceph的s3方案,具体配置看说明
# -*- coding: utf-8 -*-
import requests
import json
from email.utils import formatdate
import hmac
py3k = False
from hashlib import sha1 as sha
try:
from urlparse import urlparse
from base64 import encodestring
except:
py3k = True
from urllib.parse import urlparse
from base64 import encodebytes as encodestring
class 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', 'versions', '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-amz-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 %s:%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('x-amz-')):
interesting_headers[lk] = headers[key].strip()
# If x-amz-date is used it supersedes the date header.
if not py3k:
if 'x-amz-date' in interesting_headers:
interesting_headers['date'] = ''
else:
if 'x-amz-date' in interesting_headers:
interesting_headers['date'] = ''
buf = '%sn' % method
for key in sorted(interesting_headers.keys()):
val = interesting_headers[key]
if key.startswith('x-amz-'):
buf += '%s:%sn' % (key, val)
else:
buf += '%sn' % 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 buf
class S3Admin():
def __init__(self):
self.access_key = '' #填access_key
self.secret_key = '' #填secret_key
self.endpoint = 's3.ceph.work' #填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.headers
s3client = S3Admin()
bucket_name= 'xxx' #替换成相应的bucket名称
result = s3client.get_bucket_usage(bucket_name)
print 'objects_num= %s , total_Bytes_Used= %s ' % (result['X-RGW-Object-Count'],result['X-RGW-Bytes-Used'])
#注意 objects_num 为当前bucket内的object数量,total_Bytes_Used为当前bucket内的已用容量(单位为Byte)
- struts2 异常处理3板斧
- UE4中的单映射:TMap容器
- Visual Studio 2015 前端开发工作流
- 易学易用的Windows PowerShell
- mongodb-java-driver基本用法
- 基于Redis的开源分布式服务Codis
- 利用Weblogic的iisproxy、iisforward插件实现IIS转发
- 帆软FineReport如何使用程序数据集
- etcd:用于服务发现的键值存储系统
- 如何使用HTTP压缩优化服务器
- "org.jboss.netty.internal.LoggerConfigurator".DESCRIBED is already registered 的解决办法
- ASP.NET SignalR 高可用设计
- JavaScript 和asp.net配合编码字符串
- 3D游戏开发之UE4中的集合:TSet容器
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 变废为宝 | 旧电脑变云数据中心 | Esxi 配置
- 【Java面试总结】Java基础(上篇)
- Esxi安装LEDE(OpenWrt) | 旧电脑变庞路由
- [白话解析] 深入浅出贝叶斯定理
- LEDE(OpenWrt)安装openclash |旁路由技巧 | 配置网络流量全部通过旁路由
- [白话解析] 深入浅出朴素贝叶斯模型原理及应用
- 【Java面试总结】Java基础(下篇)
- 使用现代化的脚本进行 ArcGIS JS API 开发
- TextField suffixIcon点击时TextField 选中弹出键盘<Flutter Bug篇>
- seata redis模式重构之全局事务更新
- 【STM32F407】第7章 RL-TCPnet V7.X网络协议栈移植(RTX5)
- 【STM32F429】第7章 RL-TCPnet V7.X网络协议栈移植(RTX5)
- [白话解析] 深入浅出 极大似然估计 & 极大后验概率估计
- 【STM32H7】第7章 RL-TCPnet V7.X网络协议栈移植(RTX5)
- TinyMCE 富文本编辑器的使用实例指导