之前我们分享了操作关系型数据库，具体文章，
FastAPI 学习之路（三十二）创建数据库
这次我们分享的是非关系型数据库--Redis。
首先，我们安装对应的依赖
pip intsall aioredis==1.3.1
接下来，我们去导入创建对应的连接。
from aioredis import create_redis_pool, Redis from fastapi import FastAPI app = FastAPI() async def get_redis_pool() -> Redis: redis = await create_redis_pool(f"redis://:@127.0.0.1:6379/0?encoding=utf-8") return redis @app.on_event("startup") async def startup_event(): app.state.redis = await get_redis_pool() @app.on_event("shutdown") async def shutdown_event(): app.state.redis.close() await app.state.redis.wait_closed()
这里我们也利用了上次分享的事件，FastAPI 学习之路（五十三）startup 和 shutdown。接下来，我们去创建一个api去操作对应的一个api，进行调试。
@app.get("/test", summary="测试redis") async def test_redis(request: Request, num: int=Query(123, title="参数num")): # 等待redis写入 await异步变同步 # 如果不关心结果可以不用await，但是这里下一步要取值， # 必须得先等存完值 后再取值 await request.app.state.redis.set("test", num) # 等待 redis读取 v = await request.app.state.redis.get("test") print(v, type(v)) return {"msg": v}
我们可以用postman请求下。
我们看下。redis是否存储
我们可以看到redis存储了我们的数据，我们的接口也正常返回了。这只是一个简单的demo。后续我们可以存储缓存，也可以来存储我们的token。
