后端技术：Redis进行数据缓存的两种方法

在fastapi项目中Redis进行数据缓存的两种不同的方法的demo

第一种方法：通过FastAPI应用状态

准备文件：models/redis.py为fastapi的数据库模型文件

import os

import aioredis
from aioredis import Redis


async def sys_cache() -> Redis:
    """
    系统缓存
    :return: cache 连接池
    """

    # 由 url 方式创建 redis 连接池
    sys_cache_pool = aioredis.ConnectionPool.from_url(
        f"redis://{os.getenv('CACHE_HOST', '127.0.0.1')}:{os.getenv('CACHE_PORT', 6379)}",
        db=os.getenv('CACHE_DB', 0),
        encoding='utf-8',
        decode_responses=True
    )
    return Redis(connection_pool=sys_cache_pool)

准备文件：core/Events.py为事件监听，这startup启动函数中的app.state.cache = await sys_cache()为将上面的redis.py::sys_cache()进行注册到state中，然后进行下面的步骤。

"""
@Des: fastapi事件监听
"""

from typing import Callable
from fastapi import FastAPI
from database.mysql import register_mysql
from database.redis import sys_cache
from aioredis import Redis


def startup(app: FastAPI) -> Callable:
    """
    FastApi 启动完成事件
    :param app: FastAPI
    :return: start_app
    """
    async def app_start() -> None:
        # APP启动完成后触发
        print("启动完毕")
        # 注册数据库
        await register_mysql(app)
        # 注入缓存到app state
        app.state.cache = await sys_cache()

        pass
    return app_start


def stopping(app: FastAPI) -> Callable:
    """
    FastApi 停止事件
    :param app: FastAPI
    :return: stop_app
    """
    async def stop_app() -> None:
        # APP停止时触发
        print("停止")
        pass

    return stop_app

具体实现：

api/test_redis.py这里是方法的具体实现：

from core.Response import success
from aioredis import Redis
from database.redis import sys_cache
from fastapi import Depends, Request

# 这里就是通过FastAPI应用`state`状态实现的redis数据缓存方法
async def test_my_redis(req: Request):
    # 连接池放在request
    value = await req.app.state.cache.get("today")
    return success(data=[], msg="test_my_redis")

这种方法的特点是将 Redis 客户端保存在 FastAPI 应用的全局状态中。这样可以在整个应用中轻松访问 Redis 客户端，而不需要在每个路由函数中单独创建或传递客户端实例。且没有类型提示，需要重写FastAPI的state方法。

第二种方法：使用依赖注入

只需要对models/redis.py进行操作，下面是具体的实现：

api/test_redis.py

from core.Response import success
from aioredis import Redis
from database.redis import sys_cache
from fastapi import Depends, Request


async def test_my_redis_depends(today: int, cache: Redis = Depends(sys_cache)):
    # 连接池放在依赖注入

    # await cache.set(name="today", value=today)  # 写入库
    await cache.set(name="ex_today", value=today, ex=60)
    # value = await cache.get("today")  # 获取
    return success(data=[], msg=f"今天是{today}号")

这里对cache: Redis = Depends(sys_cache)的理解为：
cache: Redis = Depends(sys_cache) 表示使用依赖注入来获取 Redis 客户端实例。sys_cache 函数被用作依赖项，它负责创建和返回 Redis 客户端实例。

在这种方法中，依赖注入（Depends）被用于在每次请求时动态地创建或获取 Redis 客户端实例。这种方法的好处是它使得路由函数更加独立和模块化，因为它们不需要直接访问 FastAPI 应用的全局状态。有类型提示。