什么是在Python中实现与redis通信的最佳方式

Question

我正在尝试编写一个与redis通信的模块。目前为止，它正在做以下事情。

1. 获得令牌
2. 如果身份标记，然后连接到Redis的一个
3. 其他连接到Redis的乙
4. 获取数据的特定键
5. 删除键

这里是我写的：

import redis 

def get_data(token): 

    if token == "tokenA" 
     connection = redis.Redis(connection_pool=name_of_redis_engine1) 
    else: 
     connection = redis.Redis(connection_pool=name_of_redis_engine2) 

    data = connection.hgetall(token) 
    if not data: 
     raise Some Error 

    return data 

def delete_data(token): 

    connection = redis.Redis(connection_pool=name_of_redis_engine) 

    data = redis_connection.delete(token) 
    if not data: 
     raise Some Error 

    return data 

因为在两个函数中都有一些重复的数据，所以这不是一个好方法。我想知道什么是最好的方法来使它整洁可能类？ 。将欣赏任何帮助很多。

Answer 1

Thankyou @yarkee。我已经想出了以下解决方案。如果有人能指出一些更好的方法，将不胜感激。

class RedisClass(object): 
def __init__(self, token=None): 

    self.token = token 

    if self.token == "TokenA": 
     self.redis_connection = redis.Redis(connection_pool="RedisEngineA") 
    else: 
     self.redis_connection = redis.Redis(connection_pool="RedisEngineB") 

def get_data(self): 
    data = self.redis_connection.hgetall(self.token) 
    if not data: 
     raise AuthenticationError({"status: Invalid token"}, code=200, log_error=False) 
    return data 

def delete_data(self): 
    data = self.redis_connection.delete(self.token) 
    if not data: 
     raise AuthenticationError({"status: Invalid token"}, code=200, log_error=False) 
    return data 

Answer 2

让它成为一个班。

import redis 

class RedisOp(object): 
    def __init__(self): 
     self.connection = redis.Redis(connection_pool=name_of_redis_engine) 

    def get_data(self, token): 
     data = self.connection.hgetall(token) 
     if not data: 
      raise Some Error 
     return data 

    def delete_data(self, token): 
     data = self.connection.delete(token) 
     if not data: 
      raise Some Error 
     return data 


op = RedisOp() 
print op.get_data('mykey') 
op.delete_data('mykey') 

Answer 3

你可以让它成为一个类。鉴于您的具体要求，你可以做的是：

class RedisStore: 
    def __init__(self, default_connection, tokenA_connection): 
     self._default_connection = default_connection 
     self._tokenA_connection = tokenA_connection 

    def _chose_connection(token): 
     if token == "tokenA" 
      return self._tokenA_connection 
     else: 
      return self._default_connection 

    def get_data(self, token): 
     connection = self._chose_connection(token) 
     data = connection.hgetall(token) 
     if not data: 
      raise Exception("Some Error") # you can only raise exceptions, and you should use a more specific one 

     return data 

    def delete_data(self, token): 
     connection = self._chose_connection(token) 
     data = connection.delete(token) 
     if not data: 
      raise Exception("Some Error") # if that is supposed to raise the same exception, you could generalize further... 
     return data 

redis_store = new RedisStore(redis.Redis(connection_pool=name_of_redis_engine1), redis.Redis(connection_pool=name_of_redis_engine2)) 

您可以一次实例化类和重用它多次查找/删除。