多次调用asyncio.run()结合Firebase触发RuntimeError问题咨询

问题描述 #

结合Firebase异步客户端多次调用asyncio.run()时，首次上传操作成功，但第二次调用会触发RuntimeError: Event loop is closed错误。替换为普通异步操作（如字典修改）时无此异常，该问题出现在Streamlit搭建的网站中，导致仅首次表单提交有效。

简化示例代码 #

import streamlit as st
import firebase_admin
from firebase_admin import credentials, firestore_async
import json
import asyncio

key_dict = json.loads(st.secrets["textkey"])
cred = credentials.Certificate(key_dict)
app = firebase_admin.initialize_app(cred)
db_async = firestore_async.client(app)


async def _upload_async():
    dicts_to_upload = {
        "doc_1": {"key_1": "val_1"},
        "doc_2": {"key_2": "val_2"},
    }

    tasks = []

    for key in dicts_to_upload.keys():
        result_dict = db_async.collection("test-collection").document(key)
        task = result_dict.set(dicts_to_upload[key])
        tasks.append(task)

    await asyncio.gather(*tasks)


asyncio.run(_upload_async())
print("\n\nSuccess1\n\n")
asyncio.run(_upload_async())
print("\n\nSuccess2\n\n")

报错信息 #

Success1

Traceback (most recent call last):
  File "/Users/colinrichter/Documents/GitHub/YCN-Calculator/testing.py", line 100, in <module>
    asyncio.run(_upload_async())
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/Users/colinrichter/Documents/GitHub/YCN-Calculator/testing.py", line 95, in _upload_async
    await asyncio.gather(*tasks)
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/cloud/firestore_v1/async_document.py", line 131, in set
    write_results = await batch.commit(**kwargs)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/cloud/firestore_v1/async_batch.py", line 60, in commit
    commit_response = await self._client._firestore_api.commit(
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/cloud/firestore_v1/services/firestore/async_client.py", line 1055, in commit
    response = await rpc(
               ^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/api_core/retry/retry_unary_async.py", line 230, in retry_wrapped_func
    return await retry_target(
           ^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/api_core/retry/retry_unary_async.py", line 160, in retry_target
    _retry_error_helper(
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/api_core/retry/retry_base.py", line 212, in _retry_error_helper
    raise final_exc from source_exc
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/api_core/retry/retry_unary_async.py", line 155, in retry_target
    return await target()
                 ^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/api_core/timeout.py", line 120, in func_with_timeout
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/google/api_core/grpc_helpers_async.py", line 166, in error_remapped_callable
    call = callable_(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/grpc/aio/_channel.py", line 150, in __call__
    call = UnaryUnaryCall(
           ^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/grpc/aio/_call.py", line 565, in __init__
    self._invocation_task = loop.create_task(self._invoke())
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 434, in create_task
    self._check_closed()
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 519, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
sys:1: RuntimeWarning: coroutine 'UnaryUnaryCall._invoke' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

原因分析 #

这不是Firebase的BUG，而是异步客户端的资源绑定特性导致的：

• asyncio.run()每次调用都会创建新的事件循环，执行完毕后自动关闭该循环。
• Firebase异步客户端（如firestore_async.client）会将内部的连接池、GRPC异步资源绑定到首次初始化时所在的事件循环。当该循环被关闭后，后续调用时客户端尝试复用已关闭的循环，从而触发错误。
• 普通异步操作不依赖外部绑定的资源，因此可以在多次asyncio.run()调用中正常执行。

解决方案 #

方案1：单次asyncio.run()执行多次操作 #

将多次上传逻辑放在同一个异步函数中，只调用一次asyncio.run()：

async def main():
    await _upload_async()
    print("\n\nSuccess1\n\n")
    await _upload_async()
    print("\n\nSuccess2\n\n")

asyncio.run(main())

方案2：手动复用事件循环 #

手动创建并复用同一个事件循环，避免每次创建新循环：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    loop.run_until_complete(_upload_async())
    print("\n\nSuccess1\n\n")
    loop.run_until_complete(_upload_async())
    print("\n\nSuccess2\n\n")
finally:
    loop.close()

方案3：Streamlit场景适配 #

Streamlit 1.22+版本支持直接在交互回调中使用await，无需手动调用asyncio.run()，可以直接利用Streamlit自身的事件循环：

if st.button("上传数据"):
    await _upload_async()
    st.success("上传成功")

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