Fastapi의 요금 제한

Fastapi의 요금 제한

Fastapi 애플리케이션에서 API 엔드포인트 요청을 등급 제한하는 방법은 무엇인가요? 나는 사용자당 초당 5회의 API 호출 요청을 등급 제한하고 그 제한을 초과하여 해당 특정 사용자를 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)

---

패스트 API는 기본적으로 이것을 지원하지 않지만, 아래와 같은 몇 개의 라이브러리에서 가능하지만, 일반적으로 데이터베이스 백업(리디스, memcached 등)이 필요하지만, 슬로우 API는 데이터베이스가 없을 경우 메모리 폴백을 가지고 있다.

• https://pypi.org/project/fastapi-limiter/
• https://pypi.org/project/slowapi/

설명서에서 볼 수 있듯이 사용하기 위해 다음을 참조하십시오:

참고: 이 작업을 수행하려면 실행이 필요합니다.

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)

---

Fast API는 이 기능을 즉시 제공하지 않으므로 라이브러리를 사용하는 것이 가장 좋습니다.

훌륭하고, 사용하기 쉽다.

이렇게 쓰면 돼요.

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"}

---

구현하기에 매우 아름다운 패키지입니다.

그러나 사용은 또한 할 수 있지만 데이터베이스를 시작해야 한다.

1. 시작하다.

2. 파이썬 코드: 파이썬 파일 쓰기:

3. 코드:

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)

4. 이 파이썬 파일을 실행합니다.

5. 링크를 테스트하다.

---

사용할 수 있습니다

와 비교하여 사용자의 요구를 더 잘 충족시킬 수 있습니다.

예를 들어, 초당 5회 액세스 제한을 초과한 후 특정 사용자를 60초 동안 차단합니다.

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

---

몇 초 또는 몇 분 동안 필요한 데이터를 저장하기 위해 데이터베이스를 사용하는 외부 패키지를 사용하는 것보다 다음과 같이 사용하는 것이 좋습니다:

여기 클래스의 총 요청 사용 제한은 몇 초 안에 수행할 수 있습니다.

나는 클라이언트 IP를 사용하여 그들의 요청 행동을 추적하고 몇 초 이내에 초과할 경우 제한을 가하고 있다.

종속성이므로 빠른 API의 미들웨어에서는 작동하지 않을 수 있습니다(테스트하지 않았습니다)

from fastapi import FastAPI, Request, HTTPException, Depends
import time

# Initialize FastAPI app
app = FastAPI()

# In-memory storage for request counters
request_counters = {}

# Custom RateLimiter class with dynamic rate limiting values per route
class RateLimiter:
    def __init__(self, requests_limit: int, time_window: int):
        self.requests_limit = requests_limit
        self.time_window = time_window

    async def __call__(self, request: Request):
        client_ip = request.client.host
        route_path = request.url.path

        # Get the current timestamp
        current_time = int(time.time())

        # Create a unique key based on client IP and route path
        key = f"{client_ip}:{route_path}"

        # Check if client's request counter exists
        if key not in request_counters:
            request_counters[key] = {"timestamp": current_time, "count": 1}
        else:
            # Check if the time window has elapsed, reset the counter if needed
            if current_time - request_counters[key]["timestamp"] > self.time_window:
                # Reset the counter and update the timestamp
                request_counters[key]["timestamp"] = current_time
                request_counters[key]["count"] = 1
            else:
                # Check if the client has exceeded the request limit
                if request_counters[key]["count"] >= self.requests_limit:
                    raise HTTPException(status_code=429, detail="Too Many Requests")
                else:
                    request_counters[key]["count"] += 1

        # Clean up expired client data (optional)
        for k in list(request_counters.keys()):
            if current_time - request_counters[k]["timestamp"] > self.time_window:
                request_counters.pop(k)

        return True

# Include the custom RateLimiter dependency on specific routes
@app.get("/limited", dependencies=[Depends(RateLimiter(requests_limit=10, time_window=60))])
async def limited_endpoint():
    return {"message": "This endpoint has rate limiting (10 requests per 60 seconds)."}

@app.get("/limited/other", dependencies=[Depends(RateLimiter(requests_limit=5, time_window=60))])
async def limited_other_endpoint():
    return {"message": "This endpoint has rate limiting (5 requests per 60 seconds)."}

@app.get("/unlimited")
async def unlimited_endpoint():
    return {"message": "This endpoint has no rate limiting."}

우리는 사용자 정의 속도 제한 값을 매개 변수로 받아들이는 방법으로 사용자 정의 속도 제한기 클래스를 만든다. 클래스는 클래스 인스턴스를 종속 호출 가능으로 사용할 수 있는 메서드를 구현합니다.