作者：陈爱梅志杰力凯 | 来源：互联网 | 2023-09-09 02:58
如何在 Fastapi 应用程序中对 API 端点请求进行速率限制？我需要为每个用户每秒限制 API 调用 5 个请求，并且超过该限制会阻止该特定用户 60 秒。
在 main.py
def get_application() -> FastAPI:
application = FastAPI(title=PROJECT_NAME, debug=DEBUG, version=VERSION)
application.add_event_handler(
"startup", create_start_app_handler(application))
application.add_event_handler(
"shutdown", create_stop_app_handler(application))
return application
app = get_application()
在事件.py
def create_start_app_handler(app: FastAPI) -> Callable:
async def start_app() -> None:
redis = await aioredis.create_redis_pool("redis://localhost:8080")
FastAPILimiter.init(redis)
return start_app
在端点
@router.post('/user',
tags=["user"],
name="user:user", dependencies=[Depends(RateLimiter(times=5, secOnds=60))])
***code****
从此文件 test.py 运行。
import uvicorn
from app.main import app
if __name__ == "__main__":
uvicorn.run("test:app", host="0.0.0.0", port=8000, reload=True)
我如上编辑，但出现以下错误。
File "****ite-packagesstarletterouting.py", line 526, in lifespan
async for item in self.lifespan_context(app):
File "****site-packagesstarletterouting.py", line 467, in default_lifespan
await self.startup()
File "****site-packagesstarletterouting.py", line 502, in startup
await handler()
File "****appcoreservicesevents.py", line 15, in start_app
redis = await aioredis.create_redis_pool("redis://localhost:8080")
File "****site-packagesaiorediscommands__init__.py", line 188, in create_redis_pool
pool = await create_pool(address, db=db,
File "****site-packagesaioredispool.py", line 58, in create_pool
await pool._fill_free(override_min=False)
File "C****site-packagesaioredispool.py", line 383, in _fill_free
cOnn= await self._create_new_connection(self._address)
File "****site-packagesaioredisconnection.py", line 111, in create_connection
reader, writer = await asyncio.wait_for(open_connection(
File "****asynciotasks.py", line 455, in wait_for
return await fut
File "****site-packagesaioredisstream.py", line 23, in open_connection
transport, _ = await get_event_loop().create_connection(
File "****asynciobase_events.py", line 1033, in create_connection
raise OSError('Multiple exceptions: {}'.format(
OSError: Multiple exceptions: [Errno 10061] Connect call failed ('::1', 8080, 0, 0), [Errno 10061] Connect call failed ('127.0.0.1', 8080)
回答
最好的选择是使用库，因为 FastAPI 不提供此功能开箱即用。
slowapi很棒，而且易于使用。
你可以像这样使用 ut 。
from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/home")
@limiter.limit("5/minute")
async def homepage(request: Request):
return PlainTextResponse("test")
@app.get("/mars")
@limiter.limit("5/minute")
async def homepage(request: Request, response: Response):
return {"key": "value"}
京公网安备 11010802041100号