实现功能或目的
FastAPI 结合 Redis 使用方法，FastAPI 已经很快了，但是 更快的方案依然是继续结合 Redis 来当数据缓存
本文章提供3种 Redis 结合 FastAPI 的方法
分别见这里：
环境：
Windows 10
Python 3.10.11 X64
fastapi==0.108.0
aioredis==2.0.1
第1种方案
使用依赖注入来实现
实现步骤
- 安装库
aioredis>=2.0.1
dependency-injector>=4.41.0
结构：
--- Test
|-- redis.py
|-- test_router.py
|-- test_aioredis.py
- 初始化Redis连接和关闭
"""
@ File : redis.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : redis
"""
from typing import AsyncIterator
from aioredis import from_url, Redis
async def init_redis_pool(host: str, password: str, db: int = 0, port: int = 6379) -> AsyncIterator[Redis]:
session = await from_url(
url=f"redis://{host}", port=port, password=password, db=db, encoding="utf-8", decode_responses=True)
yield session
await session.close()
- 添加 server类，实现Redis 动作，Server 依赖于 Redis
"""
@ File : redis.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : redis
"""
from typing import Awaitable
from aioredis import Redis
class Service(object):
def __init__(self, redis: Redis) -> None:
self._redis = redis
async def set(self, key: str, value: str) -> Awaitable:
return await self._redis.set(key, value)
async def get(self, key: str) -> str:
return await self._redis.get(key)
- 创建一个 Container 容器，用于连接Redis并存放连接池
"""
@ File : redis.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : redis
"""
from dependency_injector import containers, providers
class Container(containers.DeclarativeContainer):
config = providers.Configuration()
redis_pool = providers.Resource(
init_redis_pool,
host=config.redis_host,
port=config.redis_port,
db=config.redis_db,
password=config.redis_password, )
service = providers.Factory(Service, redis=redis_pool)
- 配置路由，实现依赖注入
"""
@ File : test_router.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_router
"""
from fastapi import Depends, APIRouter
from dependency_injector.wiring import inject, Provide
from lib.redis import Service, Container
router = APIRouter(responses={404: {"message": "Not Found"}})
# 在需要使用缓存的位置，用 inject 装饰，并依赖注入 service: Service = Depends(Provide[Container.service])
@router.get("/router-test")
@inject
async def index(service: Service = Depends(Provide[Container.service])):
value = await service.set('name', 'value')
return {"result": value}
- 应用程序 APP
"""
@ File : test_aioredis.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis
"""
from fastapi import FastAPI, Depends
from dependency_injector.wiring import inject, Provide
from redis import Service, Container
import test_router
app = FastAPI()
app.include_router(router=test_router.router)
# 在需要使用缓存的位置，用 inject 装饰，并依赖注入 service: Service = Depends(Provide[Container.service])
@app.get("/test")
@inject
async def index(service: Service = Depends(Provide[Container.service])):
value = await service.set('name', 'value')
return {"result": value}
# 创建容器
container = Container()
container.config.redis_host.from_value("localhost")
container.config.redis_password.from_env("REDIS_PASSWORD", "<YOU PASSWORD>") # 若环境变量不存在，则使用 <YOU PASSWORD>
container.config.redis_db.from_value(0)
container.config.redis_port.from_value(6379)
# 创建容器时，要将需要依赖注入的模块加载到容器中，否则将无法调用依赖项，报属性错误
container.wire(modules=[__name__, test_router.__name__])
if __name__ == '__main__':
import uvicorn
uvicorn.run(app='test_aioredis:app', host="127.0.0.1", port=8000, reload=True)
第2种方案
使用Lifespan Events生命周期事件来实现
实现步骤
- 安装库
aioredis>=2.0.1
- 初始化Redis连接和关闭
结构：
--- Test
|-- test_router2.py
|-- test_aioredis2.py
"""
@ File : test_aioredis2.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis2
"""
from typing import AsyncIterator
from aioredis import from_url, Redis
async def init_redis_pool(host: str, password: str, db: int = 0, port: int = 6379) -> AsyncIterator[Redis]:
session = await from_url(
url=f"redis://{host}", port=port, password=password, db=db, encoding="utf-8", decode_responses=True)
return session
- 创建异步上下文管理器
"""
@ File : test_aioredis2.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis2
"""
from contextlib import asynccontextmanager
from fastapi import FastAPI
# 关于生命周期事件详见：https://fastapi.tiangolo.com/zh/advanced/events/#lifespan
@asynccontextmanager
async def lifespan(apps: FastAPI):
session = await init_redis_pool(host="127.0.0.1", password="<YOU PASSWORD>", db=0, port=6379) # 你的密码<YOU PASSWORD>
# 将Redis连接添加到app全局实例,详见：https://www.starlette.io/applications/
# Storing state on the app instance
app.state.redis = session
yield
await session.close()
- 在路由中使用APP全局实例，调用Redis
"""
@ File : test_router2.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_router2
"""
from fastapi import APIRouter, Request
router = APIRouter(responses={404: {"message": "Not Found"}})
# 利用 request 访问全局实例，若 request 可用，则request.app可用
@router.get("/router-test")
async def index(request: Request):
value = await request.app.state.redis.set('name', 'value')
return {"result": value}
- 使用 异步上下文管理器，添加到应用 APP
"""
@ File : test_aioredis2.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis2
"""
from fastapi import Request
import test_router2
app = FastAPI(lifespan=lifespan) # 添加生命周期为函数 lifespan
app.include_router(router=test_router2.router)
# 利用 request 访问全局实例，若 request 可用，则request.app可用
@app.get("/test")
async def index(request: Request):
value = await request.app.state.redis.set('name', 'value')
return {"result": value}
if __name__ == '__main__':
import uvicorn
uvicorn.run(app='test_aioredis2:app', host="127.0.0.1", port=8000, reload=True)
第3种方案
使用startup、shutdown事件来实现（不推荐，因为被Lifespan Events生命周期事件所取代）
实现步骤
- 安装库
aioredis>=2.0.1
- 初始化Redis连接和关闭
结构：
--- Test
|-- test_router3.py
|-- test_aioredis3.py
"""
@ File : test_aioredis3.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis3
"""
from typing import AsyncIterator
from aioredis import from_url, Redis
async def init_redis_pool(host: str, password: str, db: int = 0, port: int = 6379) -> AsyncIterator[Redis]:
session = await from_url(
url=f"redis://{host}", port=port, password=password, db=db, encoding="utf-8", decode_responses=True)
return session
- 创建 startup、shutdown事件
"""
@ File : test_aioredis3.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis3
"""
from fastapi import FastAPI
@app.on_event("startup")
async def startup_event():
# 将Redis连接添加到app全局实例,详见：https://www.starlette.io/applications/
# Storing state on the app instance
# 你的密码<YOU PASSWORD>
app.state.redis = await init_redis_pool(host="127.0.0.1", password="<YOU PASSWORD>", db=0, port=6379)
@app.on_event("shutdown")
async def shutdown_event():
await app.state.redis.close()
- 在路由中使用APP全局实例，调用Redis
"""
@ File : test_router3.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_router3
"""
from fastapi import APIRouter, Request
router = APIRouter(responses={404: {"message": "Not Found"}})
# 利用 request 访问全局实例，若 request 可用，则request.app可用
@router.get("/router-test")
async def index(request: Request):
value = await request.app.state.redis.set('name', 'value')
return {"result": value}
- 应用 APP，直接使用request 访问全局实例
"""
@ File : test_aioredis3.py
@ Author : yqbao
@ Version : V1.0.0
@ Description : test_aioredis3
"""
from fastapi import Request
import test_router3
app = FastAPI()
app.include_router(router=test_router3.router)
# 利用 request 访问全局实例，若 request 可用，则request.app可用
@app.get("/test")
async def index(request: Request):
value = await request.app.state.redis.set('name', 'value')
return {"result": value}
if __name__ == '__main__':
import uvicorn
uvicorn.run(app='test_aioredis3:app', host="127.0.0.1", port=8000, reload=True)