python 如何在aigram中间件中添加一个位置结构?

7xllpg7q  于 2023-06-28  发布在  Python
关注(0)|答案(1)|浏览(114)

我有几个文件。任务是配置中间件,以便在telegram bot中按下开始按钮后将用户添加到本地数据库。结果我得到一个错误。这就是:
TypeError:on_start()缺少1个必需的位置参数:'请求'
文件www.example.com

from typing import Dict, Any, Callable, Awaitable

from aiogram import BaseMiddleware
from aiogram.types import Message
from apscheduler.schedulers.asyncio import AsyncIOScheduler

class SchedulerMiddleware(BaseMiddleware):
def __init__(self, scheduler: AsyncIOScheduler):
self.scheduler = scheduler

async def __call__(self,
handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
event: Message,
data: Dict[str, Any],
) -> Any:
data['scheduler'] = self.scheduler
return await handler(event, data)

文件db_middleware. py

from typing import Callable, Awaitable, Dict, Any

from aiogram import BaseMiddleware
from aiogram.types import Message

class DbSession(BaseMiddleware):
def __init__(self, connector):
self.connector = connector

async def __call__(self,
handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
event: Message,
data: Dict[str, Any],
) -> Any:
async with self.connector.connection() as connect:
data['request'] = Request(connect)
return await handler(event, data)`

文件www.example.com

import asyncio
import logging

from aiogram import Bot, types, Dispatcher
from aiogram.filters import Command, Text
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import BotCommand, BotCommandScopeDefault
from apscheduler.schedulers.asyncio import AsyncIOScheduler

from LearnBot.middleware.middleware import SchedulerMiddleware
from LearnBot.others.db_connect import Request
from main import States

token = "5100032963:***"

class States(StatesGroup):
first_name = State()
last_name = State()
telephone = State()
complete = State()
cancel = State()

async def comm(bot: Bot):
command = [
BotCommand(
command='start',
description='Начало работы',
),
BotCommand(
command='help',
description='Помощь',
),
]

await bot.set_my_commands(command, BotCommandScopeDefault())

async def on_start(message: types.Message, state: FSMContext, request: Request):
await message.reply("Введите необходимые даные\nВведите имя " + message.text)
await request.add_user(message)
await state.set_state(States.first_name)

def register_fsm(dp: Dispatcher):
dp.message.register(cancel, States.complete, Text(text='cancel', ignore_case=True))

dp.message.register(on_start, Command(commands='start'))

async def start():
try:
logging.basicConfig(
level=logging.INFO
)

bots = Bot(token)
dp = Dispatcher()

scheduler = AsyncIOScheduler(timezone='Europe/Moscow')

# scheduler.add_job(run_interval_args, 'interval', seconds=10, args=(55,))
dp.update.middleware.register(SchedulerMiddleware(scheduler))
dp.message.register(on_start, Command(commands='start'))
scheduler.start()

await dp.start_polling(bots)
except Exception as ex:
print(ex)

if __name__ == "__main__":
asyncio.run(start())
n3h0vuf2

n3h0vuf21#

我想您忘了在文件“db_middleware.py”中传递拉取请求了

import asyncpg    
class DbSession(BaseMiddleware):
    def __init__(self, connector: asyncpg.pool.Pool):
        super().__init__()
        self.connector = connector

相关问题