在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 应用的全局状态。有类型提示。