主页 / 问题 / 1572743
Accepted
Alexey Trukhanov
Asked:2024-03-22 16:00:45 +0000 UTC2024-03-22 16:00:45 +0000 UTC

关于Postgresql异步查询、获取结果的问题

  • 772

各位同事，我对异步编程完全不懂，想搞清楚，求帮助。我编写了一个项目，使用该库psycopg2作为 Postgresql 的接口。我决定向项目添加异步，将模块更新到psycopg版本 3，支持异步，并编写了一个简单的示例来说明我的问题。

import asyncio
import psycopg
from datetime import datetime

async def main():

    async def open_connection():
        async with await psycopg.AsyncConnection.connect(
                dbname=dbname,
                password=password,
                host='localhost') as aconn:
            async with aconn.cursor() as cur:
                await cur.execute("select count(*) from table_1 join table_2 using (id)")
                print("\nВыполнили запрос к БД")
                return await cur.fetchall()

    async def print_monitor():
        while True:
            print(f'\r{datetime.now()}', end='')
            await asyncio.sleep(0)

    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(open_connection())
        task2 = tg.create_task(print_monitor())

asyncio.run(main())

在示例中，有两个协程 - 第一个查询数据库，第二个不断打印时间（只是一个时间计数器）。当数据库返回对请求的响应时，我们打印“Completed...”。

问题 （如果需要分成两个单独的问题，请写在评论中）

  1. 在同步代码中，该方法cursor.fetchall()返回基本响应（通常是元组列表）。在异步版本中，它返回一个协程。如何得到结果？
  2. 从代码中可以看到，数据库查询完成后，时间输出并不会停止。模块中asyncio或其他地方是否有特殊方法来接收协程执行的确认，并在完成后立即中断计数器。也就是说，我认为条件while应该包含某个标志，当执行对数据库的请求的协程时会触发该标志。当然，我可以引入nonlocal一个变量main()并将其用作标志，但有些东西告诉我必须有某种异步工具。

UPT

我稍微改变一下第一个问题。在所有文档和示例中都写到，在异步代码中cursor.fetchall()它应该返回结果。但对我来说，它返回一个协程，这意味着我写错了。什么？

python

1 个回答

  1. Best Answer

    使用asyncio.Event()和task.result()：

    import asyncio
import psycopg
from datetime import datetime

async def main():

    async def open_connection(event):
        async with await psycopg.AsyncConnection.connect(
                dbname=dbname,
                password=password,
                host='localhost') as aconn:
            async with aconn.cursor() as cur:
                await cur.execute("select count(*) from table_1 join table_2 using (id)")
                print("\nВыполнили запрос к БД")
                event.set()
                return await cur.fetchall()

    async def print_monitor(event):
        while not event.is_set():
            print(f'\r{datetime.now()}', end='')
            await asyncio.sleep(0)

    event = asyncio.Event()
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(open_connection(event))
        task2 = tg.create_task(print_monitor(event))
    print(task1.result())

asyncio.run(main())
    • 1

相关问题