疑惑：我现在可以通过添加“手动的关闭连接”阻止报错的发生，但这样的话就会失去“自动管理连接池”的意义。因为第一次遇见这种问题，所以请教一下大家的建议。
以下是报错提示————
True
Exception ignored in:
Traceback (most recent call last):
File "d:\Uers\创作\Ea\twoxyz\.venv\Lib\site-packages\aiomysql\connection.py", line 1131, in __del__
self.close()
File "d:\Uers\创作\Ea\twoxyz\.venv\Lib\site-packages\aiomysql\connection.py", line 339, in close
self._writer.transport.close()
File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.12_3.12.240.0_x64__qbz5n2kfra8p0\Lib\asyncio\proactor_events.py", line 109, in close
self._loop.call_soon(self._call_connection_lost, None)
File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.12_3.12.240.0_x64__qbz5n2kfra8p0\Lib\asyncio\base_events.py", line 772, in call_soon
self._check_closed()
File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.12_3.12.240.0_x64__qbz5n2kfra8p0\Lib\asyncio\base_events.py", line 519, in _check_closed
raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
这是我写的代码————
[Python] 纯文本查看 复制代码import asynciofrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import exists, select
from orm_module import QR_HTML_AccessLogs
DATABASE_URL = "mysql+aiomysql://********************************************************"
# 创建异步引擎
async_engine = create_async_engine(
DATABASE_URL,
echo=False,
)
# 定义异步数据库连接类
class AsyncDatabaseConnection:
def __init__(self):
# 创建异步会话工厂
self.AsyncSessionFactory = sessionmaker(
bind=async_engine,
class_=AsyncSession,
expire_on_commit=False
)
# 查询数据是否存在（返回布尔值）——异步
async def query_single_code(self, data_code):
async with self.AsyncSessionFactory() as db:
# 构建查询
exists_query = await db.execute(select(exists().where(QR_HTML_AccessLogs.code == data_code)))
return exists_query.scalar()
if __name__ == "__main__":
db_connection = AsyncDatabaseConnection()
# 查询单个key是否存在
bb = asyncio.run(db_connection.query_single_code("151515"))
print(bb) # 应该输出 True 或 False
