1. 安装依赖

pip install fastapi redis uvicorn

2. 主要代码

#!/usr/bin/env python
import os
import sys
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from datetime import timedelta
from pathlib import Pathfrom fastapi import FastAPI, Request
from pydantic import BaseModel
from redis import asyncio as aioredisclass RegisterRedis(AbstractAsyncContextManager):def __init__(self, app: FastAPI, *args, **kw) -> None:app.state.redis = self._redis = aioredis.Redis(*args, **kw)async def __aenter__(self) -> "RegisterRedis":await self._redis.ping()return selfasync def __aexit__(self, *args, **kw):await self._redis.aclose()@staticmethoddef get_client(request: Request) -> aioredis.Redis:return request.app.state.redis@asynccontextmanager
async def lifespan(app: FastAPI):async with RegisterRedis(app):yieldapp = FastAPI(lifespan=lifespan)@app.get("/")
async def get_all_redis_keys(request: Request) -> list[str]:redis = RegisterRedis.get_client(request)return await redis.keys()class Item(BaseModel):key: strvalue: str | int | float | bytesexpire: int | timedelta | None = None@app.post("/")
async def set_key_value(request: Request, item: Item) -> dict[str, str]:redis = RegisterRedis.get_client(request)await redis.set(item.key, item.value, item.expire or None)value = await redis.get(item.key)return {item.key: value.decode()}def runserver() -> None:"""This is for debug mode to start server. For prod, use supervisor+gunicorn instead."""import uvicorn  # type:ignoreroot_app = Path(__file__).stem + ":app"auto_reload = "PYCHARM_HOSTED" not in os.environhost = "0.0.0.0"port = 8000if sys.argv[1:]:port = int(sys.argv[1])if sys.platform == "darwin" or sys.platform.lower().startswith("win"):tool = "open" if Path("/usr/bin/open").exists() else "explorer"os.system(f"{tool} http://127.0.0.1:{port}")  # Auto open browseruvicorn.run(root_app, host=host, port=port, reload=auto_reload)if __name__ == "__main__":runserver()