Python中操作Redis的常用方法小结

brew install redis

# 开启服务
brew services start redis
# 关闭服务
brew services stop redis

% redis-cli ping
PONG
% redis-cli     
127.0.0.1:6379> set name peter
OK
127.0.0.1:6379> get name
"peter"
127.0.0.1:6379> keys *
1) "name"

pip install redis

import redis

def create_connection(host='localhost', port=6379, db=0):
    connection = None
    try:
        connection = redis.Redis(host=host, port=port, db=db)
        if connection.ping():
            print("Connection to Redis DB successful")
    except redis.ConnectionError as e:
        print(f"The error '{e}' occurred")
    
    return connection

def close_connection(connection):
    # Redis-py does not require explicit close
    print("Redis connection does not need to be closed explicitly")

# 使用示例
connection = create_connection()
close_connection(connection)

def insert_data(connection, key, value):
    try:
        connection.set(key, value)
        print(f"Data inserted: {key} -> {value}")
    except redis.RedisError as e:
        print(f"The error '{e}' occurred")

insert_data(connection, 'name', 'Alice')

def query_data(connection, key):
    try:
        value = connection.get(key)
        if value:
            print(f"Data retrieved: {key} -> {value.decode('utf-8')}")
        else:
            print(f"No data found for key: {key}")
    except redis.RedisError as e:
        print(f"The error '{e}' occurred")

query_data(connection, 'name')

def update_data(connection, key, value):
    try:
        connection.set(key, value)
        print(f"Data updated: {key} -> {value}")
    except redis.RedisError as e:
        print(f"The error '{e}' occurred")

update_data(connection, 'name', 'Bob')

def delete_data(connection, key):
    try:
        result = connection.delete(key)
        if result:
            print(f"Data deleted for key: {key}")
        else:
            print(f"No data found for key: {key}")
    except redis.RedisError as e:
        print(f"The error '{e}' occurred")

delete_data(connection, 'name')

def create_connection(host='localhost', port=6379, db=0):
    connection = None
    try:
        connection = redis.Redis(host=host, port=port, db=db)
        if connection.ping():
            print("Connection to Redis DB successful")
    except redis.ConnectionError as e:
        print("Failed to connect to Redis server")
    except redis.RedisError as e:
        print(f"Redis error: {e}")

    return connection