You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

我的Pymyq脚本出现asyncio session is closed错误是什么原因?

问题根因分析
  • pymyq库近期的版本更新调整了会话复用逻辑:库内部会隐式缓存登录产生的API实例关联的aiohttp会话,不会随你代码中创建的ClientSession销毁而同步清除缓存。你代码中每个操作函数都用async with aiohttp.ClientSession()创建会话,操作结束后上下文管理器自动关闭该会话,但第二次执行时pymyq会优先调用已被关闭的缓存会话,直接抛出Session is closed错误。
  • 你每次执行指令逻辑时都单独创建新的asyncio事件循环,旧循环关联的未完全销毁的异步资源会和新循环产生冲突,进一步放大了会话资源异常的问题。
  • 重复登录逻辑也会加剧会话冲突:你每次执行开门、关门、查询操作都重新调用login()接口,没有复用已有有效会话,多余的登录请求也会触发服务端的会话管控逻辑,增加会话被提前关闭的概率,这也是脚本此前正常、近期突然报错的核心触发原因。
修复方案

调整思路为统一管理会话和API实例,避免重复创建资源,核心修改后的代码参考如下:

import asyncio
import aiohttp
import pymyq
from pymyq.errors import MyQError

EMAIL = "替换为你的登录邮箱"
PASSWORD = "替换为你的登录密码"
DEVICE_ID = "替换为你的车库门设备ID"

# 全局缓存API实例和会话,避免重复创建
api = None
websession = None

async def init_device():
    global api, websession
    # 仅在会话关闭或未初始化时创建新会话
    if not websession or websession.closed:
        websession = aiohttp.ClientSession()
    # 仅在API未初始化时调用登录接口
    if not api:
        api = await pymyq.login(EMAIL, PASSWORD, websession)
    return api.devices[DEVICE_ID]

async def door_position():
    try:
        device = await init_device()
        return device.state
    except MyQError as err:
        return 'error'

async def door_open():
    try:
        device = await init_device()
        await device.open()
        return 'success'
    except MyQError as err:
        return 'error'

async def door_close():
    try:
        device = await init_device()
        await device.close()
        return 'success'
    except MyQError as err:
        return 'error'

# 复用同一个事件循环,不要重复新建
loop = asyncio.get_event_loop()

def handle_command(command):
    Input = command.lower()
    try:
        if 'garage door status' in Input:
            door_status = loop.run_until_complete(door_position())
            return f'The garage door is {door_status}'
        elif 'open the garage door' in Input:
            loop.run_until_complete(door_open())
            return 'The garage door is now opening'
        elif 'close the garage door' in Input:
            loop.run_until_complete(door_close())
            return 'The garage door is now closing'
        else:
            return 'command unknown.'
    except Exception as e:
        return f'*******ERROR*****{str(e)}'

内容的提问来源于stack exchange,提问作者Erik Caldwell

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.30 21:57:04