创建 psycopg2 包装函数以仅调用函数一次来执行

Question

我正在尝试创建函数 psycopg2 游标执行包装函数。

我就是这样尝试的。

async def query_execute(query, *args):
    connection = psycopg2.connect(**db_params) # This db_params already declare before.
    cursor = connection.cursor()
    try:
        cursor.execute(query, *args)
        return cursor.fetchone()
    finally:
        connection.close()

这段代码有很多问题。

• 这个游标只能做fetchone。
• connection.close 可能无法执行。

我的目标就是这样

result = await query_execute("SELECT * From users where id = $1 AND email = $2", ("test","test"))
result_one = result.fetchone()

请告诉我如何修复我的代码。

Answer 1

也许这就是我的解决方案。

async def query_execute(query, *args):
    connection = await asyncpg.connect(**db_params)
    try:
        result = await connection.fetch(query, *args)
        return result
    except Exception as e:
        print(f"Error:{e}")
        raise
    finally:
        await connection.close()