Tair (Redis® OSS-Compatible):Gestion des connexions multi-appareils avec TairHash

# -*- coding: utf-8 -*-
#!/usr/bin/env python
import os
import time
from tair import Tair
from tair import ResponseError

def get_tair() -> Tair:
    # Connect to a Tair instance.
    # host: endpoint of your Tair instance
    # port: default is 6379
    # username: leave blank to use the default account
    # password: read from environment variable TAIR_PASSWORD
    return Tair(
        host="r-bp************.redis.rds.aliyuncs.com",
        port=6379,
        db=0,
        username="",
        password=os.environ.get("TAIR_PASSWORD"),
    )

def add_user_token(tair: Tair, user_id: str, device: str, token: str, ttl_seconds: int) -> bool:
    # Store a session token using EXHSET with a per-field TTL.
    # key   = user_id  (groups all devices for one user)
    # field = device   (identifies the device type)
    # value = token    (the session token)
    # ex    = TTL in seconds
    try:
        result = tair.exhset(user_id, device, token, ex=ttl_seconds)
        return result == 1
    except ResponseError as e:
        print(e)
        return False

def print_active_tokens(tair: Tair, user_id: str):
    # EXHGETALL returns only fields that have not yet expired.
    for entry in tair.exhgetall(user_id):
        print("{}:{}".format(user_id, entry))

if __name__ == "__main__":
    tair = get_tair()

    user_1 = "user1"
    user_2 = "user2"

    # Write four tokens with different TTLs.
    add_user_token(tair, user_1, "phone", "token_123", 5)   # expires in 5 s
    add_user_token(tair, user_1, "pad",   "token_124", 10)  # expires in 10 s
    add_user_token(tair, user_2, "pad",   "token_456", 10)  # expires in 10 s
    add_user_token(tair, user_2, "pc",    "token_457", 10)  # expires in 10 s

    # Wait 6 seconds. user1's phone token (TTL = 5 s) will have expired.
    print("Waiting 6 seconds...")
    time.sleep(6)

    # Only 3 of the 4 tokens are returned; user1/phone has expired.
    print_active_tokens(tair, user_1)  # prints user1/pad only
    print_active_tokens(tair, user_2)  # prints user2/pad and user2/pc

Waiting 6 seconds...
user1:{field: pad, value: token_124}
user2:{field: pad, value: token_456}
user2:{field: pc, value: token_457}