如何在aiogram 3.11处理器间传递类对象？含Middleware报错排查

错误原因分析 #

你遇到的TypeError是因为aiogram 3.x的处理器采用依赖注入机制，不能直接将data: Dict[str, Any]作为处理器参数声明。处理器会自动从data字典中解析匹配名称的参数，而非接收整个字典。

最优实现方案 #

通过中间件将自定义类实例注入到事件的data字典，然后在处理器中直接声明对应参数即可完成传递。以下是完整示例：

1. 编写通用用户中间件 #

这个中间件会自动为所有包含from_user的事件（如Message、CallbackQuery）注入TgUser实例：

from typing import Callable, Dict, Any, Awaitable
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject, Message, CallbackQuery
from your_module import UsersTable, TgUser  # 替换为你的实际模块路径

class UserMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
        event: TelegramObject,
        data: Dict[str, Any]
    ) -> Any:
        # 判断事件是否包含用户信息
        if isinstance(event, (Message, CallbackQuery)) and event.from_user:
            user_data = UsersTable.get_or_none(UsersTable.tg_id == event.from_user.id)
            if user_data:
                # 封装为自定义TgUser实例
                tg_user = TgUser(
                    tg_id=user_data.tg_id,
                    name=user_data.name,
                    surname=user_data.surname,
                    username=user_data.username,
                    confirm_code=user_data.confirm_code,
                    is_attached=user_data.is_attached
                )
                data["tg_user"] = tg_user
            else:
                # 用户不存在时注入None，可根据业务逻辑调整
                data["tg_user"] = None
        return await handler(event, data)

2. 注册中间件 #

将中间件注册到Dispatcher，可选择全局注册（所有事件生效）或针对特定事件类型注册：

from aiogram import Dispatcher

dp = Dispatcher()

# 全局注册：所有更新事件都会经过该中间件
dp.update.middleware(UserMiddleware())

# 或仅针对消息事件注册
# dp.message.middleware(UserMiddleware())

3. 在处理器中使用自定义实例 #

处理器直接声明tg_user: TgUser参数即可，aiogram会自动从data中注入：

from aiogram.filters import StateFilter
from aiogram.types import Message
from aiogram.fsm.context import FSMContext
from your_module import TgUser, State  # 替换为你的实际模块路径

@dp.message(StateFilter(State.START))
async def handler_button_activate(
    message: Message,
    state: FSMContext,
    tg_user: TgUser | None  # 加上None处理用户不存在的场景
):
    if tg_user:
        await message.answer(f"欢迎回来，{tg_user.name}！")
        # 直接使用tg_user实例的属性和方法
        print(tg_user.is_attached)
    else:
        await message.answer("请先完成注册！")

额外说明 #

• 若要传递其他自定义类实例，只需在中间件中向data字典添加对应键值对，再在处理器中声明同名参数即可。
• 如需限制中间件仅对特定事件生效，可改为注册到dp.callback_query.middleware()或dp.edited_message.middleware()等。
• 对于需要持久化的非实时数据，可结合FSMContext的存储功能，但中间件+数据库查询的方式更适合实时同步用户数据。

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