异步SQLAlchemy创建数据库记录失败，报'NoneType'无'send'属性错误

使用异步SQLAlchemy向数据库插入用户记录时失败，提交会话时触发AttributeError: 'NoneType' object has no attribute 'send'，同时伴随事件循环已关闭的错误。

数据库连接代码 #

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from contextlib import asynccontextmanager

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/asyncalchemystudy"

engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=0)
Base = declarative_base()
async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False
)


@asynccontextmanager
async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session


async def init_models():
    async with engine.begin() as conn:
        # await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

SQLAlchemy模型 #

class users(Base):
    __tablename__ = "users"

    password = Column(TEXT)
    username = Column(TEXT)
    id = Column(Integer, primary_key=True)

创建用户的代码 #

async def create_user(session: AsyncSession, username: text, password: text) -> tasks | None:
    user = users(username=username, password=password)
    try:
        session.add(task)
        await session.commit() <- problem here
        await session.refresh(user)
        return task
    except Exception as ex:
        print(ex)
        await session.rollback()

后续使用代码 #

创建用户后会在aiohttp会话中使用：

async with get_session() as sql_session:
        task = await queries.create_user(sql_session, username, password)
        async with aiohttp.ClientSession() as session:

异常信息 #

提交时触发异常：

AttributeError: 'NoneType' object has no attribute 'send'

回溯信息 #

Traceback (most recent call last):
  File "C:\Python\Lib\asyncio\base_events.py", line 761, in call_soon
    self._check_closed()
  File "C:\Python\Lib\asyncio\base_events.py", line 519, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Users\user\PycharmProjects\pythonProject\main.py", line 68, in <module>
    asyncio.run(telegrator_parser())
  File "C:\Python\Lib\asyncio\runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\asyncio\base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Users\user\PycharmProjects\pythonProject\main.py", line 40, in telegrator_parser
    task = await queries.create_task(sql_session)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\PycharmProjects\pythonProject\sql\queries.py", line 23, in create_task
    raise ex
  File "C:\Users\user\PycharmProjects\pythonProject\sql\queries.py", line 18, in create_task
    await session.commit()
  File "C:\Python\Lib\site-packages\sqlalchemy\ext\asyncio\session.py", line 578, in commit
    return await greenlet_spawn(self.sync_session.commit)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 134, in greenlet_spawn
    result = context.throw(*sys.exc_info())
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\session.py", line 1431, in commit
    self._transaction.commit(_to_root=self.future)
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\session.py", line 829, in commit
    self._prepare_impl()
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\session.py", line 808, in _prepare_impl
    self.session.flush()
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\session.py", line 3363, in flush
    self._flush(objects)
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\session.py", line 3502, in _flush
    with util.safe_reraise():
  File "C:\Python\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "C:\Python\Lib\site-packages\sqlalchemy\util\compat.py", line 207, in raise_
    raise exception
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\session.py", line 3463, in _flush
    flush_context.execute()
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\unitofwork.py", line 456, in execute
    rec.execute(self)
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\unitofwork.py", line 630, in execute
    util.preloaded.orm_persistence.save_obj(
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\persistence.py", line 244, in save_obj
    _emit_insert_statements(
  File "C:\Python\Lib\site-packages\sqlalchemy\orm\persistence.py", line 1237, in _emit_insert_statements
    result = connection._execute_20(
             ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\engine\base.py", line 1620, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\sql\elements.py", line 325, in _execute_on_connection
    return connection._execute_clauseelement(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\engine\base.py", line 1487, in _execute_clauseelement
    ret = self._execute_context(
          ^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\engine\base.py", line 1851, in _execute_context
    self._handle_dbapi_exception(
  File "C:\Python\Lib\site-packages\sqlalchemy\engine\base.py", line 2036, in _handle_dbapi_exception
    util.raise_(exc_info[1], with_traceback=exc_info[2])
  File "C:\Python\Lib\site-packages\sqlalchemy\util\compat.py", line 207, in raise_
    raise exception
  File "C:\Python\Lib\site-packages\sqlalchemy\engine\base.py", line 1808, in _execute_context
    self.dialect.do_execute(
  File "C:\Python\Lib\site-packages\sqlalchemy\engine\default.py", line 732, in do_execute
    cursor.execute(statement, parameters)
  File "C:\Python\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 479, in execute
    self._adapt_connection.await_(
  File "C:\Python\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 76, in await_only
    return current.driver.switch(awaitable)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 129, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "C:\Python\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 408, in _prepare_and_execute
    await adapt_connection._start_transaction()
  File "C:\Python\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 716, in _start_transaction
    self._handle_exception(error)
  File "C:\Python\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 684, in _handle_exception
    raise error
  File "C:\Python\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 714, in _start_transaction
    await self._transaction.start()
  File "C:\Python\Lib\site-packages\asyncpg\transaction.py", line 138, in start
    await self._connection.execute(query)
  File "C:\Python\Lib\site-packages\asyncpg\connection.py", line 317, in execute
    return await self._protocol.query(query, timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "asyncpg\protocol\protocol.pyx", line 338, in query
  File "asyncpg\protocol\protocol.pyx", line 331, in asyncpg.protocol.protocol.BaseProtocol.query
  File "asyncpg\protocol\coreproto.pyx", line 1078, in asyncpg.protocol.protocol.CoreProtocol._simple_query
  File "asyncpg\protocol\protocol.pyx", line 929, in asyncpg.protocol.protocol.BaseProtocol._write
  File "C:\Python\Lib\asyncio\proactor_events.py", line 365, in write
    self._loop_writing(data=bytes(data))
  File "C:\Python\Lib\asyncio\proactor_events.py", line 401, in _loop_writing
    self._write_fut = self._loop._proactor.send(self._sock, data)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'send'

Process finished with exit code 1

问题分析与修复方案 #

1. 变量名与类型标注错误 #

create_user函数存在多处低级错误：

• 创建的对象是user，但错误使用未定义的task变量进行添加和返回
• 参数类型标注使用了未定义的text，应改为str
• 返回类型标注错误，应匹配users模型而非tasks

修复后的代码：

async def create_user(session: AsyncSession, username: str, password: str) -> users | None:
    user = users(username=username, password=password)
    try:
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user
    except Exception as ex:
        print(ex)
        await session.rollback()
        return None  # 明确返回None，避免隐式返回空值

2. 事件循环关闭问题 #

回溯中显示RuntimeError('Event loop is closed')，说明数据库操作时事件循环已被提前关闭。需确保所有异步操作（包括数据库操作和aiohttp请求）都在事件循环运行期间完成：

• 确保asyncio.run包裹的主函数内所有异步任务都被正确await
• 避免在数据库会话的上下文管理器外执行依赖数据库的操作，比如将aiohttp请求放在get_session的上下文内：

async with get_session() as sql_session:
    user = await queries.create_user(sql_session, username, password)
    async with aiohttp.ClientSession() as session:
        # 这里执行aiohttp请求，确保在sql_session关闭前完成必要操作

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