标签：pipeline 打印 取数据 count version 清空 members cores ==
实际代码只有一点，其他的为备忘
# -*- coding:utf-8 -*-
import MySQLdb
import redis
class Config:
def __init__(self):
self.mysql_host = ‘192.168.44.60‘
self.mysql_user = ‘root‘
self.mysql_port = 3306
self.mysql_password = ‘123456‘
self.mysql_db = ‘bamboo‘
self.mysql_charset = ‘utf8‘
self.mysql_show_tables = "show tables"
self.mysql_desc_table = "desc %s"
self.mysql_query_info = "select * from %s"
self.redis_host = "192.168.44.60"
self.redis_port = 6379
self.redis_passwd = 123456
self.redis_db = 0
config = Config()
class MySQLHelper:
myVersion = 0.1
def __init__(self):
self.host = config.mysql_host
self.user = config.mysql_user
self.password = config.mysql_password
self.charset = config.mysql_charset
self.db = config.mysql_db
try:
self.conn = MySQLdb.connect(host=self.host, user=self.user, passwd=self.password,db=self.db,charset=self.charset)
self.cursor = self.conn.cursor()
except MySQLdb.Error as e:
print (‘MySql Error : %d %s‘ % (e.args[0], e.args[1]))
def query(self, sql):
try:
cursor = self.cursor
cursor.execute(sql)
return cursor
except MySQLdb.Error as e:
print(‘MySql Error: %s SQL: %s‘ % (e, sql))
def queryOnlyRow(self, sql):
try:
cursor = self.query(sql)
result = cursor.fetchall()
desc = cursor.description
row = {}
for i in range(0, len(result)):
row[desc[i][0]] = result[i]
return row
except MySQLdb.Error as e:
print(‘MySql Error: %s SQL: %s‘ % (e, sql))
def queryAll(self, sql):
try:
cursor = self.query(sql)
result = cursor.fetchall()
desc = cursor.description
rows = []
for cloumn in result:
row = {}
for i in range(0, len(cloumn)):
row[desc[i][0]] = cloumn[i]
rows.append(row)
return rows
except MySQLdb.Error as e:
print(‘MySql Error: %s SQL: %s‘ % (e, sql))
def queryTableNums(self,sql):
try:
cursor = self.query(sql)
result = cursor.fetchall()
row = []
for cloumn in result:
row.append(cloumn[0])
return row
except MySQLdb.Error as e:
print(‘Mysql Error: %s SQL: %s‘ %(e,sql))
def insert(self, tableName, pData):
try:
newData = {}
for key in pData:
newData[key] = "‘" + str(pData[key]) + "‘"
key = ‘,‘.join(newData.keys())
value = ‘,‘.join(map(str,newData.values()))
sql = "insert into " + tableName + "(" + key + ") values(" + value + ")"
self.query("set names ‘utf8‘")
self.query(sql)
self.commit()
except MySQLdb.Error as e:
self.conn.rollback()
print(‘MySql Error: %s %s‘ % (e.args[0], e.args[1]))
def update(self, tableName, pData, whereData):
try:
newData = []
keys = pData.keys()
for i in keys:
item = "%s=%s" % (i, "‘" + pData[i] + "‘")
newData.append(item)
items = ‘,‘.join(newData)
newData2 = []
keys = whereData.keys()
for i in keys:
item = "%s=%s" % (i, "‘" + whereData[i] + "‘")
newData2.append(item)
whereItems = " AND ".join(newData2)
sql = "update " + tableName + " set " + items + " where " + whereItems
self.query("set names ‘utf8‘")
self.query(sql)
self.commit()
except MySQLdb.Error as e:
self.conn.rollback()
print(‘MySql Error: %s %s‘ % (e.args[0], e.args[1]))
def getLastInsertRowId(self):
return self.cursor.lastrowid
def getRowCount(self):
return self.cursor.rowcount
def commit(self):
self.conn.commit()
def close(self):
self.cursor.close()
self.conn.close()
class RedisHelper():
__redis_instance = None
def __init__(self):
self.host = config.redis_host
self.port = config.redis_port
self.passwd = config.redis_passwd
self.db = config.redis_db
try:
self.connPoor = redis.ConnectionPool(host=self.host,port=self.port,db=self.db,password=self.passwd)
self.rd = redis.Redis(connection_pool=self.connPoor)
except redis.DataError as e:
print(‘Redis Error : %d %s‘ % (e.args[0], e.args[1]))
@staticmethod
def getObj():
if RedisHelper.__redis_instance:
return RedisHelper.__redis_instance
else:
RedisHelper.__redis_instance = RedisHelper().rd
return RedisHelper.__redis_instance
# string 操作
def setkey(self,key,value,**kwargs):
"""
set(name, value, ex=None, px=None, nx=False, xx=False)
:param key: 键
:param value: 值
:param kwargs:
在Redis中设置值，默认，不存在则创建，存在则修改
参数：
ex，过期时间（秒）
px，过期时间（毫秒）
nx，如果设置为True，则只有name不存在时，当前set操作才执行
xx，如果设置为True，则只有name存在时，当前set操作才执行
:return: True or False
"""
res = self.rd.set(key,value,**kwargs)
return res
def getkey(self,key):
return self.rd.get(key)
def setnxkey(self,key,value):
"""
:param key:
:param value:
:return: True or False
设置值，只有key不存在时，执行设置操作（添加）
"""
return self.rd.setnx(key,value)
def setexkey(self,key,value,time=5):
"""
:param key:
:param value:
:param time: 过期时间 (秒 或 timedelta对象)
:return: True or False
"""
return self.rd.setex(key,time,value)
def psetexkey(self,key,value,time):
"""
:param key:
:param value:
:param time: time_ms，过期时间（数字毫秒 或 timedelta对象）
:return:
"""
return self.rd.psetex(key,time,value)
def msetkey(self,mapping):
"""
批量设置多个值
:param mapping: {‘k1‘:‘v1‘,‘k2‘:‘v2‘} or k1=‘v1‘,k2=‘v2‘
:return:
"""
return self.rd.mset(mapping)
def mgetkey(self,keys,*args):
"""
批量获取多个值
‘k1‘,‘k2‘ or [‘k1‘,‘k2‘]
:param keys:
:param args:
:return:
"""
return self.rd.mget(keys,args)
def getsetkey(self,key,value):
"""
获取值并设置新值
:param key:
:param value:
:return:
"""
return self.rd.getset(key,value)
def getrangekey(self,key,start,end):
"""
获取子序列（根据字节获取，非字符） 一个汉字，是占3个字节
:param key:
:param start: bytes 起始位置(包含)
:param end: bytes 结束位置（包含）
:return:
set ‘name‘ ‘helloworld‘
OK
GETRANGE ‘name‘ 0 2
"hel"
getrange ‘name‘ 0 -1
"helloworld"
"""
return self.rd.getrange(key,start,end)
def setrange(self,key,index,value):
"""
修改字符串内容，从指定字符串索引开始向后替换（新值太长时，则向后添加）
参数：
:param key:
:param index: 字符串的索引，字节（一个汉字三个字节）
:param value: 要设置的值
:return:
"""
return self.rd.setrange(key,index,value)
def strlenkey(self,key):
"""
返回name对应值的字节长度（一个汉字3个字节）
:param key:
:return:
"""
return self.rd.strlen(key)
def incrkey(self,key):