使用 asyncpg 读取 PostgreSQL 表

Question

我知道如何使用 psycopg2、sqlalchemy、dask 读取远程服务器中的 PostgreSQL 表，但我对读取表的处理时间并不满意，并开始研究更快的替代方案，我发现 asyncpg 比其他方案快 7 倍，但 asyncpg 的文档非常丰富与上面提到的库相比很差，那里有很多例子。

我的问题是：如何高效读取 PostgreSQL 表？

我尝试过如下：

import asyncio
import asyncpg
import pandas as pd

from sshtunnel import SSHTunnelForwarder #Allow connection with SSH like PuttY connection
from sshtunnel import SSHTunnelForwarder, create_logger #Allow to follow the processes running

SSHTunnelForwarder(('IP_detail', Port_number),
        ssh_private_key=r'path_to_the_ssh_key_in_my_computer',

        ssh_username="username",
        #ssh_password="password", 
        remote_bind_address=('localhost', port_number),
        local_bind_address=('localhost', port_number),
        logger=create_logger(loglevel=1) #Makes processes being ran displayed
                           )

conn = await asyncpg.connect(user='username', password='password',
                                 database='database_name', host='127.0.0.1', port='port')


values = await conn.fetch('''SELECT * FROM table_name''')

values=pd.DataFrame(values)
values

通过上面的代码，我得到了 PostgreSQL 表中每列的所有行值，但不显示列名称，它显示列编号而不是它们的正确名称。 如何纠正这个问题？

Answer 1

使用

dict(values)

查看记录和负载的键值对

Answer 2

首先，提取您的列名称：

columns = [c.name for c in values.get_attributes()]

然后，创建您的数据框：

values = pd.DataFrame(values, columns=columns)

参见 https://github.com/MagicStack/asyncpg/issues/173#issuecomment-538055841

Answer 3

hellycopterinjuneer 的答案中的 link 是正确的，但答案并不表明有必要创建准备好的语句。为了方便起见，我在这里报告了链接中的完整代码。

async def fetch_as_dataframe(conn: asyncpg.Connection, query: str, *args):
    stmt = await conn.prepare(query)
    columns = [a.name for a in stmt.get_attributes()]
    data = await stmt.fetch(*args)
    return pd.DataFrame(data, columns=columns)

Answer 4

它在我的 FastApi 应用程序中适用于我，无需 get_attributes():

values = await app.state.db.fetch("SELECT * FROM ... ")
df = DataFrame([dict(row) for row in data])