Telethon中contextvars跨事件传递变量失败的问题求助

错误信息 #

ERROR - Unhandled exception on show_selected_tags
Traceback (most recent call last):
File "/home/kozorez/.cache/pypoetry/virtualenvs/content-telegram-bot-4LA9pChh-py3.10/lib/python3.10/site-packages/telethon/client/updates.py", line 570, in _dispatch_update
await callback(event)
File "/mnt/c/Users/kozor/PycharmProjects/content-telegram-bot/bot/bot.py", line 77, in show_selected_tags
some_var = tags_cv.get()
LookupError: <ContextVar name='tags' at 0x7f7e3e222de0>

问题描述 #

使用Telethon开发Telegram机器人时，尝试将send_tag_list()函数中通过contextvars.ContextVar设置的tags_cv值传递到show_selected_tags()函数，但触发上述LookupError。推测原因是Telethon的事件回调运行在不同的异步上下文，导致ContextVar无法跨任务共享数据。

相关代码如下：

bot.py #

"""Логика взаимодействия с ботом"""

import logging
import contextvars
from telethon.sync import TelegramClient, events

from services.channel import get_channel_data, add_channel_to_db
from services.post import parse_posts
from services.tag import get_tags_list
from services.keyboard import create_tags_keyboard

from config import (
    api_id,
    api_hash,
    bot_token
)


bot = TelegramClient(
    'bot',
    api_id,
    api_hash).start(bot_token=bot_token)


channel_data_cv = contextvars.ContextVar('channel_data')
tags_cv = contextvars.ContextVar('tags')


@bot.on(events.NewMessage(pattern='/start'))
async def send_welcome(event) -> None:
    """Отправка приветственного сообщения"""

    await event.reply('Привет! Я — бот, который поможет тебе составить пост с навигацией по твоему телеграм-каналу. \
                      \nПросто пришли мне ссылку на свой телеграм-канал.')
    

@bot.on(events.NewMessage(pattern='https://t\.me/(\S+)'))
async def send_tag_list(event) -> None:
    """Анализируем посты из канала и возвращаем клавиатуру с тегами"""

    channel_link = event.text
    tags = None

    try:
        channel_data = await get_channel_data(channel_link)
        channel_data_cv.set(channel_data)
    except ValueError as error:
        if 'Cannot get entity from a channel' in str(error):
            await event.reply('Канал должен быть публичным')
        logging.error(error)

    try:
        await add_channel_to_db(channel_data)
    except Exception as error:
        logging.error(error)
        
    try:     
        await event.reply('Посты анализируются, ожидайте')
        await parse_posts(channel_link, channel_data)
    except ValueError as error:
        logging.error(error)
        await event.reply('К сожалению, мне не удалось найти ни одного тега')

    try:
        tags = await get_tags_list(channel_data)
        tags_cv.set(tags)
    except Exception as error:
        logging.error(error)

    tags_keyboard = await create_tags_keyboard(tags)

    await event.respond("Выберите теги", buttons=tags_keyboard)


@bot.on(events.CallbackQuery())
async def show_selected_tags(event) -> None:
    some_var = tags_cv.get()
    print(some_var)

run.py #

"""Запуск бота"""

import asyncio
import logging

from bot import bot

from db.models import *

from config import (
    engine,
    client
    )



loop = asyncio.get_event_loop()
asyncio.set_event_loop(loop)



async def init_db() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
        logging.info('База данных инициализирована')


async def start() -> None:
    await init_db()

    async with bot:
        await bot.run_until_disconnected()
        logging.info('Бот запущен')

    async with client:
        await client.run_until_disconnected()
        logging.info('Клиент запущен')


if __name__ == "__main__":
    loop.create_task(start())
    loop.run_forever()

问题原因 #

Telethon的每个事件回调（比如send_tag_list和show_selected_tags）是独立的异步任务，不属于同一个上下文链。ContextVar仅能在同一个协程或其派生的子协程中共享数据，无法跨独立任务传递值，因此会触发LookupError。

解决方法 #

方法1：用用户/聊天ID关联全局字典（推荐） #

利用用户的chat_id作为唯一标识，将数据存储在全局字典中，实现跨事件共享：

# 替换原contextvars定义
user_data = {}  # 结构: {chat_id: {"tags": [], "channel_data": {}}}

# 修改send_tag_list函数存储数据
async def send_tag_list(event) -> None:
    # ... 原有代码 ...
    try:
        tags = await get_tags_list(channel_data)
        # 用当前聊天ID作为键存储数据
        user_data[event.chat_id] = {
            "tags": tags,
            "channel_data": channel_data
        }
    except Exception as error:
        logging.error(error)

# 修改show_selected_tags函数获取数据
async def show_selected_tags(event) -> None:
    chat_id = event.chat_id
    if chat_id not in user_data:
        await event.respond("请先发送频道链接")
        return
    some_var = user_data[chat_id]["tags"]
    print(some_var)

注意：多用户场景下可添加过期清理逻辑，避免内存溢出。

方法2：数据库临时存储 #

将用户临时数据存入数据库，添加user_session表，字段包含user_id、tags、channel_data、expire_time。在send_tag_list中写入数据，show_selected_tags中读取，定时清理过期记录，适合需要持久化数据的场景。

方法3：缓存服务（如Redis） #

分布式部署场景下，用Redis以用户ID为键存储临时数据，并设置过期时间，避免数据堆积，同时支持多实例共享状态。

方法4：嵌入数据到回调按钮Payload #

将需要传递的数据编码为JSON字符串，作为回调按钮的data参数，在show_selected_tags中解析：

# 修改create_tags_keyboard函数
async def create_tags_keyboard(tags):
    import json
    from telethon.tl.types import KeyboardButtonCallback
    buttons = []
    for tag in tags:
        # 将标签和全量tags数据编码为JSON
        payload = json.dumps({"action": "select_tag", "tag": tag, "all_tags": tags})
        buttons.append([KeyboardButtonCallback(text=tag, data=payload.encode())])
    return buttons

# 修改show_selected_tags函数
async def show_selected_tags(event) -> None:
    import json
    try:
        payload = json.loads(event.data.decode())
        some_var = payload["all_tags"]
        print(some_var)
    except json.JSONDecodeError:
        await event.respond("无效操作")

此方案无需依赖全局状态，适合数据量较小的场景。

内容的提问来源于stack exchange，提问作者Kozorez-V