封装 Redis 连接可以让你的应用程序代码更加整洁，同时便于管理和维护。以下是如何在 Python 中封装 Redis 连接的一个示例，特别适用于 FastAPI 应用程序。我们将创建一个 Redis 客户端类来管理连接，并确保其易于配置和使用。

1. 创建 Redis 客户端封装

首先，我们创建一个 Redis 客户端的封装类，该类负责初始化连接并提供一些基本操作方法（如设置、获取键值）。

Python

import redis
from typing import Optional, Any

class RedisClient:
    def __init__(self, host: str = 'localhost', port: int = 6379, password: Optional[str] = None, db: int = 0):
        """
        初始化 Redis 客户端。
        
        :param host: Redis 服务器地址，默认为 localhost。
        :param port: Redis 服务器端口，默认为 6379。
        :param password: Redis 密码，默认为 None。
        :param db: 使用的数据库编号，默认为 0。
        """
        try:
            self.client = redis.Redis(host=host, port=port, password=password, db=db, decode_responses=True)
            # 测试连接
            if not self.client.ping():
                raise Exception("无法连接到 Redis 服务器")
            print("Redis 连接成功")
        except Exception as e:
            print(f"Redis 连接失败: {e}")
            raise

    def set_key(self, key: str, value: Any) -> bool:
        """
        设置 Redis 键值对。
        
        :param key: 键名。
        :param value: 值。
        :return: 操作是否成功。
        """
        try:
            return self.client.set(key, value)
        except Exception as e:
            print(f"设置键值对时出错: {e}")
            return False

    def get_key(self, key: str) -> Optional[Any]:
        """
        获取 Redis 键对应的值。
        
        :param key: 键名。
        :return: 键对应的值或 None 如果键不存在。
        """
        try:
            return self.client.get(key)
        except Exception as e:
            print(f"获取键值时出错: {e}")
            return None

    def delete_key(self, key: str) -> bool:
        """
        删除 Redis 中的键。
        
        :param key: 要删除的键名。
        :return: 操作是否成功。
        """
        try:
            return self.client.delete(key) > 0
        except Exception as e:
            print(f"删除键时出错: {e}")
            return False

2. 在 FastAPI 中使用 Redis 客户端

接下来，在 FastAPI 应用中使用这个封装好的 Redis 客户端。我们可以将其作为依赖项注入到需要的地方。

Python

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# 初始化 Redis 客户端
redis_client = RedisClient(host="your_redis_host", port=6379, password="your_redis_password", db=0)

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI with Redis!"}

@app.post("/set/{key}/{value}")
def set_key_value(key: str, value: str, redis_conn: RedisClient = Depends(lambda: redis_client)):
    success = redis_conn.set_key(key, value)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to set key-value pair.")
    return {"message": f"Set {key} to {value}"}

@app.get("/get/{key}")
def get_key_value(key: str, redis_conn: RedisClient = Depends(lambda: redis_client)):
    value = redis_conn.get_key(key)
    if value is None:
        return {"message": "Key not found"}
    else:
        return {"key": key, "value": value}

@app.delete("/delete/{key}")
def delete_key(key: str, redis_conn: RedisClient = Depends(lambda: redis_client)):
    success = redis_conn.delete_key(key)
    if not success:
        raise HTTPException(status_code=404, detail="Key not found or failed to delete.")
    return {"message": f"Deleted {key}"}

3. 运行应用

确保你已经安装了所有必要的库：

Bash

pip install fastapi uvicorn redis

然后运行你的 FastAPI 应用：

Bash

uvicorn your_script_name:app --reload

将 your_script_name 替换为你保存上述代码的文件名（不包括 .py 扩展名）。例如，如果你的文件名为 main.py，则命令应为：

Bash

uvicorn main:app --reload

总结

RedisClient 类：封装了 Redis 的连接逻辑以及常用的操作方法（设置、获取、删除键值对），使得代码更易于维护。
FastAPI 集成：通过 FastAPI 的依赖注入机制 (Depends) 将 Redis 客户端实例传递给各个路由处理函数，这样可以保持 Redis 连接的一致性和高效性。
错误处理：每个操作都包含了基本的异常处理，确保即使出现错误也能给客户端一个清晰的反馈。
这种设计不仅提高了代码的可读性和可维护性，还方便了后续功能的扩展和优化。如果需要进一步的功能（如事务支持、发布/订阅模式等），也可以在这个基础上进行扩展。