fastapi-limiter更新至v0.2.0后与之前不太不一样，这里以接入Redis进行示例说明
安装：
pipinstallredis[hiredis]==7.4.0 pipinstallfastapi-limiter创建 Redis 管理器
def_get_redis_pool(use_cache_db:bool=True):uri=settings.REDIS_CACHE_URIifuse_cache_dbelsesettings.REDIS_URIreturnredis.ConnectionPool.from_url(uri.encoded_string(),decode_responses=True)redis_client=redis.Redis(connection_pool=_get_redis_pool(False))redis_cache_client=redis.Redis(connection_pool=_get_redis_pool())注意这里我使用了一个参数来区别 Redis 的 DB，实际上这里的URI来自pydantic，示例如下：
@computed_field@propertydefREDIS_URI(self)->RedisDsn:returnRedisDsn.build(scheme="redis",host=self.REDIS_HOST,port=self.REDIS_PORT,password=self.REDIS_PASSWORDorNone,path=str(self.REDIS_DB))@computed_field@propertydefREDIS_CACHE_URI(self)->RedisDsn:returnRedisDsn.build(scheme="redis",host=self.REDIS_HOST,port=self.REDIS_PORT,password=self.REDIS_PASSWORDorNone,path=str(self.REDIS_CACHE_DB))创建pyrate_limiter限制器并使用 Redis
因为fastapi-limiter基于pyrate-limiter实现，这里我们需要先创建限制器，然后提供给fastapi_limiter
frompyrate_limiterimportRate,Duration,Limiter,RedisBucket# 用于区分全局和业务的不同限制_GLOBAL_BUCKET_KEY="global-rate-limit"_BIZ_BUCKET_KEY="biz-rate-limit"# 自定义默认限流_DEFAULT_RATES=[Rate(60,Duration.MINUTE),# 60 次每分钟Rate(60*5,Duration.MINUTE*10),# 300 次每十分钟Rate(1000,Duration.HOUR)# 1000 次每小时]defget_redis_limiter(rates:list[Rate]|None=None,is_global:bool=False)->Limiter:ifis_global:bucket_key=_GLOBAL_BUCKET_KEYifnotrates:rates=_DEFAULT_RATESelse:ifnotrates:raiseValueError("Biz limit rate rules cannot be None")bucket_key=_BIZ_BUCKET_KEY# 使用我们已有 redis 管理器中的客户端bucket=RedisBucket.init(rates,redis_client,bucket_key)returnLimiter(bucket)依赖注入方法
这里可在前置路由依赖中先校验 jwt 后将用户编号或者 ID 放到request.state中，随后依赖该get_rate_limiter方法
须实现用户唯一标识才可用于判断是否登录并用来实现精确流控
fromfastapiimportRequestfromfastapi_limiter.dependsimportRateLimiterfromfastapi_limiter.identifierimportdefault_identifierfrompyrate_limiterimportRateasyncdefget_identifier(request:Request)->str:# 自定义实现逻辑user_no=getattr(request.state,"user_no",None)ifuser_no:returnf"user_{user_no}:{request.scope["path"]}"returnawaitdefault_identifier(request)defget_rate_limiter(rules:list[Rate]|None=None,is_global:bool=False)->RateLimiter:returnRateLimiter(get_redis_limiter(rules,is_global),get_identifier)路由全局限流
此处示例，假设get_current_user方法中实现了对request.state的用户唯一身份设置，就能轻松区分是否已登录用户并分别限流
get_current_user中应当实现对jwt的解码，拿到用户唯一编号后进行数据库查询，确认无误后设置到request.state（参考逻辑）
fromfastapiimportAPIRouter,Dependsfromapp.api.depsimportget_current_userfromapp.infrastructure.limiter.api_rate_limiterimportget_rate_limiter api=APIRouter(dependencies=[Depends(get_current_user),Depends(get_rate_limiter(is_global=True))])路由单独限流
由于FastAPI的dependencies是叠加的，所以上方RedisBucket的key需要区分
否则在例如发送短信等接口上会因为全局限流已经写入了一条 Redis 记录，导致设置 1 分钟 1 次的限流不可用，会直接触发429请求频繁
这里例如我们有一个发送短信接口：
_SEND_CODE_RATES=[Rate(1,Duration.MINUTE),# 1 次每分钟Rate(5,Duration.HOUR),# 5 次每小时Rate(10,Duration.DAY)# 10 次每天]@router.post("/send-code",dependencies=[Depends(get_rate_limiter(_SEND_CODE_RATES))])asyncdefsend_code():pass在触发一次短信发送后，会往 Redis 写入两条记录如下：
ZSET Key: global-rate-limit Value: user_123456:/api/send-code:8:1:adee7ca88b:1 ZSET Key: biz-rate-limit Value: user_123456:/api/send-code:8:2:7d2ecd023e:1
