如何优化Aiogram中角色过滤器的重复数据库查询问题？

核心思路：通过中间件提前获取用户角色，避免重复查询 #

在请求处理流程的最前端（中间件）一次性查询用户角色并缓存到请求上下文中，后续所有过滤器直接复用该缓存值，无需重复访问数据库。

步骤1：编写用户角色中间件 #

这个中间件会在每个消息请求进来时，先查询一次用户角色，并存入请求上下文的data字典中：

from aiogram import BaseMiddleware
from aiogram.types import Message
from typing import Callable, Dict, Any, Awaitable
# 如果使用同步ORM（如SQLAlchemy同步版），需要导入run_sync避免阻塞事件循环
from aiogram.utils.asyncio_helper import run_sync

class UserRoleMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: Dict[str, Any]
    ) -> Any:
        # 异步ORM直接用await查询
        # user = await User.query.filter_by(telegram_id=event.from_user.id).first()
        
        # 同步ORM用run_sync包装执行
        user = await run_sync(User.query.filter_by(telegram_id=event.from_user.id).first)()
        
        # 将角色存入上下文，无匹配用户则设为None
        data["user_role"] = user.role if user else None
        return await handler(event, data)

步骤2：注册中间件到Dispatcher #

将中间件绑定到消息处理流程：

dp.message.middleware(UserRoleMiddleware())

步骤3：修改RoleFilter复用上下文角色 #

修改原过滤器，不再单独查询数据库，直接从上下文读取已缓存的角色：

from aiogram.dispatcher.filters import BoundFilter
from aiogram.types import Message
from typing import Dict, Any

class RoleFilter(BoundFilter):
    key = 'is_role'
    required = False

    def __init__(self, is_role):
        self.is_role = is_role

    async def check(self, msg: Message, data: Dict[str, Any]):
        # 从上下文获取提前缓存的角色
        user_role = data.get("user_role")
        # 验证角色是否匹配
        return user_role == self.is_role

dp.filters_factory.bind(RoleFilter)

额外优化建议 #

• 如果角色变更不频繁，可以结合内存缓存（如异步版lru_cache）或Redis缓存用户角色，进一步减少数据库查询次数，但需注意在角色变更时及时清理缓存。
• 若部分请求不需要角色验证，可在中间件中通过检查处理器的flags判断是否需要查询角色，避免无意义的数据库访问（适合高并发场景）。

内容的提问来源于stack exchange，提问作者makcim512