class UserCache(UserMixin): REDIS_PREFIX = f"{PROJECT_NAME}-user" def __init__(self, alternative_id, userid=None, username=None, homepage=None, permissions=None): self.alternative_id = alternative_id self.userid = userid self.username = username self.homepage = homepage self.permissions = permissions or [] def get_id(self): return self.alternative_id def add_to_cache(self): values = { "alternative_id": self.alternative_id, "userid": self.userid, "username": self.username, "homepage": self.homepage, "permissions": ",".join(self.permissions), } redis_key = f"{self.REDIS_PREFIX}:{self.alternative_id}" redis_client.hmset(redis_key, values) redis_client.expire(redis_key, SIGNIN_LIFETIME) def del_user_cache(self): redis_client.delete(f"{self.REDIS_PREFIX}:{self.alternative_id}") @classmethod def load_from_cache(cls, alternative_id): redis_key = f"{cls.REDIS_PREFIX}:{alternative_id}" user_cache_dict = redis_client.hgetall(redis_key) if user_cache_dict: user_cache_dict['permissions'] = user_cache_dict.get('permissions', '').split(",") return cls(**user_cache_dict) 优化并提出建议,python3.8
时间: 2024-02-16 07:05:02 浏览: 111
Redis-begin.zip_Redis java_adodb REDIS_java redis_redis_site:www
以下是对代码的一些优化和建议:
1. 在类定义中,将 REDIS_PREFIX 定义为类变量,而不是实例变量。这样可以提高代码的可读性,并且只需要在类定义中定义一次。
2. 使用 typing 模块来声明函数和方法的参数和返回值,以提高代码的可读性和可维护性。
3. 在 load_from_cache 方法中,使用 redis_client.hgetall 返回的字典对象的 items 方法来遍历键值对,而不是使用 for 循环和 get 方法。这样可以在一行代码中创建 user_cache_dict。
4. 使用 redis 的 pipeline 方法来优化 add_to_cache 方法,以减少与 redis 的通信次数。
优化后的代码如下所示:
```
from typing import List
class UserCache(UserMixin):
REDIS_PREFIX: str = f"{PROJECT_NAME}-user"
def __init__(self, alternative_id: str, userid: str = None, username: str = None, homepage: str = None, permissions: List[str] = None):
self.alternative_id = alternative_id
self.userid = userid
self.username = username
self.homepage = homepage
self.permissions = permissions or []
def get_id(self) -> str:
return self.alternative_id
def add_to_cache(self) -> None:
values = {
"alternative_id": self.alternative_id,
"userid": self.userid,
"username": self.username,
"homepage": self.homepage,
"permissions": ",".join(self.permissions),
}
redis_key = f"{self.REDIS_PREFIX}:{self.alternative_id}"
with redis_client.pipeline() as pipe:
pipe.hmset(redis_key, values)
pipe.expire(redis_key, SIGNIN_LIFETIME)
pipe.execute()
def del_user_cache(self) -> None:
redis_client.delete(f"{self.REDIS_PREFIX}:{self.alternative_id}")
@classmethod
def load_from_cache(cls, alternative_id: str) -> 'UserCache':
redis_key = f"{cls.REDIS_PREFIX}:{alternative_id}"
user_cache_dict = dict(redis_client.hgetall(redis_key).items())
if user_cache_dict:
user_cache_dict['permissions'] = user_cache_dict.get('permissions', '').split(",")
return cls(**user_cache_dict)
```
阅读全文