TTL、PTTL：获取键的剩余生存时间

redis> SET song_title "Rise up, Rhythmetal"
OK
redis> TTL song_title
(integer) -1
redis> PTTL song_title
(integer) -1

redis> TTL not_exists_key
(integer) -2
redis> PTTL not_exists_key
(integer) -2

redis> TTL msg
(integer) 0

import random
from hashlib import sha256

# 会话的默认过期时间
DEFAULT_TIMEOUT = 3600*24*30    # 一个月

# 会话状态
SESSION_NOT_LOGIN_OR_EXPIRED = "SESSION_NOT_LOGIN_OR_EXPIRED"
SESSION_TOKEN_CORRECT = "SESSION_TOKEN_CORRECT"
SESSION_TOKEN_INCORRECT = "SESSION_TOKEN_INCORRECT"

def generate_token():
    """
    生成一个随机的会话令牌。
    """
    random_string = str(random.getrandbits(256)).encode('utf-8')
    return sha256(random_string).hexdigest()


class LoginSession:

    def __init__(self, client, user_id):
        self.client = client
        self.user_id = user_id
        self.key = "user::{0}::token".format(user_id)

    def create(self, timeout=DEFAULT_TIMEOUT):
        """
        创建新的登录会话并返回会话令牌，
        可选的 timeout 参数用于指定会话的过期时间（以秒为单位）。
        """
        # 生成会话令牌
        token = generate_token()
        # 储存令牌，并为其设置过期时间
        self.client.set(self.key, token, ex=timeout)
        # 返回令牌
        return token

    def validate(self, input_token):
        """
        根据给定的令牌验证用户身份。
        这个方法有三个可能的返回值，分别对应三种不同情况：
        1. SESSION_NOT_LOGIN_OR_EXPIRED —— 用户尚未登录或者令牌已过期
        2. SESSION_TOKEN_CORRECT —— 用户已登录，并且给定令牌与用户令牌相匹配
        3. SESSION_TOKEN_INCORRECT —— 用户已登录，但给定令牌与用户令牌不匹配
        """
        # 获取用户令牌
        user_token = self.client.get(self.key)
        # 令牌不存在
        if user_token is None:
            return SESSION_NOT_LOGIN_OR_EXPIRED
        # 令牌存在并且未过期，那么检查它与给定令牌是否一致
        if input_token == user_token:
            return SESSION_TOKEN_CORRECT
        else:
            return SESSION_TOKEN_INCORRECT

    def destroy(self):
        """
        销毁会话。
        """
        self.client.delete(self.key)

>>> from redis import Redis
>>> from login_session import LoginSession
>>> client = Redis(decode_responses=True)
>>> uid = "peter"
>>> session = LoginSession(client, uid) # 创建会话
>>> token = session.create() # 创建令牌
>>> token
'89e77eb856a3383bb8718286802d32f6d40e135c08dedcccd143a5e8ba335d44'
>>> session.validate("wrong token") # 验证令牌
'SESSION_TOKEN_INCORRECT'
>>> session.validate(token)
'SESSION_TOKEN_CORRECT'
>>> session.destroy() # 销毁令牌
>>> session.validate(token) # 令牌已不存在
'SESSION_NOT_LOGIN_OR_EXPIRED'

>>> token = session.create(timeout=3) # 创建有效期为3s的令牌
>>> session.validate(token) # 3s内访问
'SESSION_TOKEN_CORRECT'
>>> session.validate(token) # 超过3s之后，令牌已被自动销毁
'SESSION_NOT_LOGIN_OR_EXPIRED'

class AutoComplete:

    def __init__(self, client):
        self.client = client

    def feed(self, content, weight=1, timeout=None):
        """
        根据用户输入的内容构建自动补全结果，
        其中 content 参数为内容本身，而可选的 weight 参数则用于指定内容的权重值，
        至于可选的 timeout 参数则用于指定自动补全结果的保存时长（单位为秒）。
        """
        for i in range(1, len(content)):
            key = "auto_complete::" + content[:i]
            self.client.zincrby(key, weight, content)
            if timeout is not None:
                self.client.expire(key, timeout)  # 设置/更新键的生存时间

    def hint(self, prefix, count):
        """
        根据给定的前缀 prefix ，获取 count 个自动补全结果。
        """
        key = "auto_complete::" + prefix
        return self.client.zrevrange(key, 0, count-1)

>>> from redis import Redis
>>> from auto_complete import AutoComplete
>>> client = Redis(decode_responses=True)
>>> ac = AutoComplete(client)
>>> ac.feed("Redis", timeout=10); ac.feed("Coffee", timeout=10) # 同时执行两个调用

>>> ac.feed("Redis", timeout=10)

>>> ac.hint("Re", 10)
['Redis']
>>> ac.hint("Co", 10)
[]