Fastapi 中的速率限制

Question

如何在 Fastapi 应用程序中对 API 端点请求进行速率限制？我需要 ratelimit 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()

在events.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-packages\starlette\routing.py", line 526, in lifespan
    async for item in self.lifespan_context(app):
  File "****site-packages\starlette\routing.py", line 467, in default_lifespan
    await self.startup()
  File "****site-packages\starlette\routing.py", line 502, in startup
    await handler()
  File "****app\core\services\events.py", line 15, in start_app
    redis = await aioredis.create_redis_pool("redis://localhost:8080")
  File "****\site-packages\aioredis\commands\__init__.py", line 188, in create_redis_pool
    pool = await create_pool(address, db=db,
  File "****site-packages\aioredis\pool.py", line 58, in create_pool
    await pool._fill_free(override_min=False)
  File "C****\site-packages\aioredis\pool.py", line 383, in _fill_free
    conn = await self._create_new_connection(self._address)
  File "****site-packages\aioredis\connection.py", line 111, in create_connection
    reader, writer = await asyncio.wait_for(open_connection(
  File "****\asyncio\tasks.py", line 455, in wait_for
    return await fut
  File "****\site-packages\aioredis\stream.py", line 23, in open_connection
    transport, _ = await get_event_loop().create_connection(
  File "****\asyncio\base_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)

Answer 1

FastAPI 本身不支持这一点，但有一些库可以实现，例如下面的库，但通常需要某种数据库支持（redis、memcached 等），尽管 slowapi 有内存回退以防万一没有数据库。

为了使用 fastapi-limiter，如他们的文档所示：

注意：您需要运行 Redis 才能正常工作。

import aioredis
import uvicorn
from fastapi import Depends, FastAPI

from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter

app = FastAPI()


@app.on_event("startup")
async def startup():
    redis = await aioredis.create_redis_pool("redis://localhost")
    FastAPILimiter.init(redis)


@app.get("/", dependencies=[Depends(RateLimiter(times=2, seconds=5))])
async def index():
    return {"msg": "Hello World"}


if __name__ == "__main__":
    uvicorn.run("main:app", debug=True, reload=True)

Answer 2

最好的选择是使用库，因为 FastAPI 不提供现成的此功能。

slowapi 很棒，而且易于使用。

你可以这样使用ut。

from fastapi import FastAPI
from slowapi.errors import RateLimitExceeded
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"}

Answer 3

fastapi-limiter 和 slowapi 是非常漂亮的包来实现 Ratelimit in Fastapi。

但是用walrus也可以。但应该启动 redis 数据库。

开始redis。
python代码：写一个python文件：code1228.py
代码：

from walrus import Database, RateLimitException
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn

db = Database()
rate = db.rate_limit('xxx', limit=5, per=60)  # in 60s just can only click 5 times

app = FastAPI()


@app.exception_handler(RateLimitException)
def parse_rate_litmit_exception(request: Request, exc: RateLimitException):
    msg = {'success': False, 'msg': f'please have a tea for sleep, your ip is: {request.client.host}.'}
    return JSONResponse(status_code=429, content=msg)


@app.get('/')
def index():
    return {'success': True}


@app.get('/important_api')
@rate.rate_limited(lambda request: request.client.host)
def query_important_data(request: Request):
    data = 'important data'
    return {'success': True, 'data': data}


if __name__ == "__main__":
    uvicorn.run("code1228:app", debug=True, reload=True)

运行这个 python 文件。
测试 link.http://127.0.0.1:8000/important_api

Answer 4

您可以使用https://github.com/abersheeran/asgi-ratelimit

相比https://pypi.org/project/fastapi-limiter/ and https://pypi.org/project/slowapi/，更能满足您的需求。

示例：超过每秒五次访问限制后，阻止特定用户 60 秒。

app.add_middleware(
    RateLimitMiddleware,
    authenticate=AUTH_FUNCTION,
    backend=RedisBackend(),
    config={
        r"^/user": [Rule(second=5, block_time=60)],
    },
)

Fastapi 中的速率限制

Ratelimit in Fastapi

python

rate-limiting

fastapi