項目背景：
之前在做項目的時候，由于是采用微服務架構，所有服務通信使用
Redis
作為資料互動層。需求是不同類型的資料存儲在不同的資料庫中，不同的資料庫就需要動态切換
Redis
資料庫。
Redis
預設有
16
個資料庫，
index
(值範圍：
0~15
)，預設
index=0
。切換資料庫指令為：
select index
。
但是呢，
flask_redis
卻沒有這個功能。為什麼會沒有呢？因為
redis-py
就沒有。
flask_redis
是基于
redis-py
進行二次封裝的庫，
redis-py
為了保證
Redis
執行個體的線程安全，沒有實作
SELECT
指令。
Redis 用戶端執行個體可以安全地線上程間共享。從内部實作來說，隻有在指令執行時才擷取連接配接執行個體，完成後直接傳回連接配接池，指令永不修改用戶端執行個體的狀态。但是，有一點需要注意：SELECT 指令。SELECT 指令允許切換目前連接配接使用的資料庫。新的資料庫保持被選中狀态，直到選中另一個資料庫或連接配接關閉。這會導緻在傳回連接配接池時，連接配接可能指定了别的資料庫。是以，redis-py 沒有在用戶端執行個體中實作 SELECT 指令。如果要在同一個應用中使用多個 Redis 資料庫，應該給第一個資料庫建立獨立的用戶端執行個體（可能也需要獨立的連接配接池）。
參考：https://segmentfault.com/q/1010000012594548
解決方案
flask_redis
既然不能動态切換資料庫，那我們就從根源入手——使用前都重新連接配接并指定資料庫。
感覺是挺暴力的，其實，這是官方建議的做法。自己動手豐衣足食，接下來我們就來封裝一個支援動态切換資料庫的
Redis
，需要滿足使用者在使用的時候無感覺。
廢話不多說，直接上代碼：
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import copy
import datetime
from redis import StrictRedis, ConnectionPool
class RedisLib(object):
def __init__(self, redis_db, redis_url, blacklist=None):
self.redis = None
self.redis_db = redis_db
self.redis_url = redis_url
self.blacklist = blacklist
self.blacklist_data = None
def select(self, db):
url = '%s/%s' % (self.redis_url, db.split('db')[1])
pool = ConnectionPool.from_url(url=url, decode_responses=True)
self.redis = StrictRedis(connection_pool=pool)
def push_redis(self, db, data):
def handle_data():
self.blacklist_data = [value for value in map(
lambda index: data.pop(index) if data.get(index) else None, self.blacklist)]
key = '%s:%s' % (self.redis_db[db], data['id'])
for k, v in data.items():
self.redis.hset(key, k, v.strftime("%Y-%m-%d %H:%M:%S") if isinstance(
v, datetime.datetime) else (v if v else ''))
self.select(db)
if isinstance(data, list):
for obj in data:
data = copy.deepcopy(obj.__dict__)
handle_data()
else:
data = copy.deepcopy(data.__dict__)
handle_data()
def pull_redis(self, db, _id=None):
self.select(db)
key = '%s:%s' % (self.redis_db[db], _id if _id else '')
if _id is None:
data = self.redis.dbsize()
elif _id == 'key':
data = self.redis.keys()
elif _id == '*':
data = [self.redis.hgetall(key) for key in self.redis.keys()]
else:
data = self.redis.hgetall(key)
return data
def __del__(self):
self.redis.connection_pool.disconnect()
以上代碼就是實作了新增資料、擷取資料、切換資料庫的功能，在執行個體化的時候需要進行配置。雖然可以使用了，但是并不友好，因為我每次都得需要指定資料庫，那有沒有什麼方法可以連資料庫指定都不用呢？
當然有的啦！
那就是再寫一個[狗頭]哈哈！
class HandleQueue(RedisLib):
def __init__(self):
self.redis_db = {
'db0': None,
'db1': 'oss:aop:user',
'db9': 'oss:aop:role'
}
self.redis_url = 'redis://127.0.0.1:6379'
# Flask app config from redis url
# from app import app
# self.redis_url = app.config['REDIS_URL']
self.blacklist = ['_sa_instance_state', 'version', 'status']
RedisLib.__init__(self, self.redis_db, self.redis_url, self.blacklist)
def set_user_data(self, data):
return self.push_redis('db1', data)
def get_user_data(self, user_id=None):
return self.pull_redis('db1', user_id)
def set_role_data(self, data):
return self.push_redis('db9', data)
def get_role_data(self, dict_id=None):
return self.pull_redis('db9', dict_id)
if __name__ == '__main__':
handle_queue = HandleQueue()
return_value = handle_queue.get_user_data(1)
print(return_value)
運作結果：
{'username': '極客點兒', 'role_id': '2', 'phone': '', 'account': 'GeekDot', 'mail': '', 'update_time': '2020-09-22 15:52:32', 'create_time': '2020-08-13 16:51:51', 'token': 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MDA4NDc1NTIsImlhdCI6MTYwMDc2MTE1MiwiaXNzIjoiMjF2aWFuZXQiLCJkYXRhIjp7ImlkIjoxfX0.fEAfuRC3tvA-qv5j6Xjbqu9W8ksY4PZwYwFoI3ST3xU', 'number': '', 'id': '1'}
通過類繼承把切換資料庫等基本功都保留，然後通過構造函數配置資料庫，最後用到哪些資料和資料庫就寫一個很簡單的方法即可。
因為我們是
Flask
項目，是以要使用
Flask
的配置檔案，方法如下，也很簡單。
# Flask app config from redis url
# from app import app
# self.redis_url = app.config['REDIS_URL']
因為項目組中用的是
ORM
是以
push_redis
現在隻能接受
ORM
對象。如果想支援其他對象，可以自己改寫，也簡單。
最後别忘了使用析構函數将資料庫關閉！
![【MySQL資料庫】資料庫索引事務1.索引2.事務[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)