将来の置き換えを見据えた、FastAPI用レートリミッターのミニマム設計
FastAPIでレートリミットを自作する理由と最低限の実装
FastAPIを利用してアプリケーションを実装している、かつアプリケーション層で、レートリミットを実装したい場合、最低限の機能を備えたレートリミットを自作するか、ライブラリを採用するか迷ったので、記事を書きます。
結論、要求される要件が少ないのであれば、拡張性を担保することは前提として、自作でも良いと考えています。ただ、個人的には車輪の再発明は出来る限り避けたく、良いものが世の中にあるのであればそちらを使いたいので、エコシステムの習熟を期待しています。
レートリミットとは
Webアプリケーションにおいて、リクエストの過剰な集中を防ぐために、レートリミットの導入は重要な対策の一つです。APIに対して不正に高頻度なアクセスが発生した場合、システムに負荷をかけたり、他のユーザーに対してサービス品質の低下を招く恐れがあります。
以下のコードは、IPアドレスごとに、60秒あたり最大5リクエストまで許可する 固定ウィンドウ方式のミドルウェアの実装例です。ストレージにはメモリとRedisの両方をサポートしており、引数で切り替え可能な構造になっています。
ライブラリの状況
レートリミットのためのライブラリも存在していて、slowapiや、fastapi-limiterを利用することで、簡単にレートリミット機能を導入できます。
しかし、2025年4月の時点では、これらのライブラリは活発にメンテナンスされておらず、Pull Request や Issue にも対応の遅れが見られます。Issueを覗いてみると、継続的にメンテナンスを行うコアなメンバーが不足していることがわかります。
そのような状況下で、ライブラリを導入するコストが高いため、自分で最低限のレートリミット機構を実装する必要性が出てきました。ただし、あくまで「今は自作」であっても、将来的に優れたライブラリが登場した際にはスムーズに置き換えられるように、設計や構造には一定の抽象化を意識する必要があると考えます。
自作を選んだ理由
ライブラリの状況に加えて、今回作成するアプリケーションでは要件が比較的シンプルでした。具体的には、「すべてのAPIエンドポイントに共通で、一定回数以上のアクセスを制限する」という非常に単純なレートリミットの要件でした。ユーザーごとの制御や、動的なルール変更といった高度な機能は、少なくとも初期段階では必要ありません。
また、アプリケーションの規模もまだ小さく、関与するメンバーも少数です。こういった条件下では、最初は初期の要件を満たすように自作実装を行い、良いものが出たときに乗り換える、という手段に一定合理性が出てくると考えました。
最小の要件整理
今回のレートリミット実装では、最小限かつ現実的な機能に絞って設計しています。要件は以下のとおりです。
- すべてのAPIに共通で適用できる
アプリケーション全体に対して一律の制限をかけられる設計にする。window(時間幅)と limit(許可回数)の設定が可能。 - IPアドレス単位で制限する
クライアントのIPアドレスをキーとして、個別にレートリミットを適用。 - 過剰アクセス時はHTTP 429(Too Many Requests)を返す
アクセス上限を超えた場合は、標準に則って 429 ステータスコードを返却 - アルゴリズムは Fixed Window を採用
シンプルな時間固定ウィンドウ方式を採用。将来的に切り替えられるように、アルゴリズム部分は抽象化 - ストレージ選択可能にする
開発初期やスモールスケールな運用ではメモリで十分。スケーラブルにしたい場合に備え、Redisベースのストレージにも差し替え可能な設計にする
実装のイメージ
まずは、最小構成で動作するレートリミットの実装のイメージを示します。FastAPIではリクエストをハンドリングする前に処理を挟むために、Middleware を使うのが最も簡単な方法です。
以下のコードは、IPアドレスごとに、60秒あたり最大5リクエストまで許可する 固定ウィンドウ方式のミドルウェア実装のイメージです。
# main.py
limiter = MyRatelimiter(limit=5, window=60)
app.add_middleware(RateLimitMiddleware, limiter=limiter)
# limiter middleware
class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app, limiter: IRateLimiter):
super().__init__(app)
self.limiter = limiter
async def dispatch(self, request, call_next):
if not self.limiter.is_allowed(request):
raise HTTPException(status_code=429, detail="Too many requests")
return await call_next(request)
# limiter class
from abc import ABC, abstractmethod
class IRateLimiter(ABC):
@abstractmethod
def is_allowed(self, key: str) -> bool:
pass
class MyRatelimiter(IRateLimiter):
def __init__(self, limit: int, window: int, storage: RateLimitStorage, strategy: RateLimitstorategy):
self.limit = limit
self.window = window
self.storage = storage or InMemoryStorage()
self.strategy = strategy or FixedWindowRateLimiter()
def is_allowed(self, request) -> bool:
return self.strategy.check_request(request, self.storage)
抽象化
自作でレートリミットを実装するとはいえ、今後の拡張性やライブラリ置き換えを前提にしておくことが重要です。今回の実装では、「Middleware」「RateLimiter」「Storage」「Strategy」の責務を明確に分離し、インターフェースによる抽象化を行っています。
RateLimiterのインターフェース設計
IRateLimiter は、リクエストが制限を超えているかどうかを判定する共通のインターフェースです。これにより、異なるアルゴリズムやストレージ方式の実装を共通の型で扱えるようになります。
class IRateLimiter(ABC):
@abstractmethod
def is_allowed(self, request) -> bool:
pass
具体的な制御ロジックは MyRatelimiter の中に実装されており、アルゴリズム(Strategy)とストレージ(Storage)も依存性注入(DI)で外部から切り替え可能にしています。
Middlewareとの分離
レートリミットのミドルウェア(RateLimitMiddleware)は、Limiterの中身に依存せず、単に is_allowed() を呼び出すだけの構造になっています。これにより、RateLimiter本体の実装を変えてもMiddleware側のコードは変更不要です。
if not self.limiter.is_allowed(request):
raise HTTPException(status_code=429)
この構造は、将来的に別のライブラリをラップする形で導入する際にも、インターフェースを満たすクラスを作成するだけで済む、というメリットがあります。
DI(依存性注入)による実装差し替えの柔軟性
初期実装では InMemoryStorage × FixedWindowStrategy のような組み合わせで始め、負荷が増えたら RedisStorage や SlidingWindowStrategy へ切り替える、といった拡張が非常にスムーズです。
MyRatelimiter(
limit=5,
window=60,
storage=RedisStorage(),
strategy=SlidingWindowRateLimiter()
)
このように、シンプルな構造ながら柔軟に拡張可能な設計にしておくことで、現状のニーズを満たしつつ、将来的に有望なライブラリへの乗り換えなどが容易になります。
まとめ
本記事では、FastAPIでレートリミットを実装するにあたって、既存ライブラリの将来性や要件のシンプルさを踏まえ、自作を選択した経緯と、その最小実装の構造について紹介しました。
今回の対応はあくまで暫定的な対応だと考えていて、将来的に成熟したライブラリが登場するようであれば(もしすでに存在しているのであれば、誰か教えてください)、それに切り替えていこうと考えています。
Discussion