Dynamic Redis Database Switching in Flask with a Custom Wrapper

When building a micro‑service project that uses Redis as the data‑exchange layer, the need often arises to store different types of data in separate Redis databases and switch between them dynamically. The standard flask_redis extension, which is a thin wrapper around redis‑py , does not implement the SELECT command because it aims to keep the Redis client instance thread‑safe.

The Redis client can safely be shared across threads, but the SELECT command changes the selected database for the underlying connection, which may cause subsequent operations to use the wrong database when the connection is returned to the pool. Therefore, redis‑py deliberately omits SELECT from the client API.

To work around this limitation, the article proposes creating a custom Redis wrapper that reconnects to the required database each time a request is made. The solution includes a RedisLib class that handles connection pool creation, dynamic select , data insertion ( push_redis ), and data retrieval ( pull_redis ), as well as a destructor that disconnects the pool.

#!/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()

To make the wrapper easy to use in a Flask project, a subclass HandleQueue is defined that configures the database mapping, Redis URL, and blacklist, and provides simple methods such as set_user_data , get_user_data , set_role_data , and get_role_data .

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)

The script prints a dictionary containing user information retrieved from Redis, demonstrating that the custom wrapper successfully switches databases and performs CRUD operations without requiring the caller to manage connections manually.

Finally, the article reminds developers to close the Redis connection pool in the destructor to avoid resource leaks.