SQLAlchemy - 如何从 ResultProxy 访问列名并写入 CSV 标题

Question

我正在尝试使用 SQLAlchemy 建立与 PostgreSQL 数据库的连接，执行 SQL 查询并将文件的输出打印到 linux 中的文件。

from sqlalchemy import create_engine
import yaml
import csv

outfile = open('/path/filename.csv', 'wb')
outcsv = csv.writer(outfile, delimiter='\t')


with open('config.yml') as f:
    cfg = yaml.safe_load(f)
    username = cfg['credentials']['username']
    password = cfg['credentials']['password']
    host = cfg['credentials']['host'] 
    port = cfg['credentials']['port']
    dbname = cfg['credentials']['dbname']
    engine = create_engine('postgresql://{}:{}@{}:{}/{}'.format(username, password, host, port, dbname))
    result = engine.execute("""select * from db.tablename """)

    # dump column titles (optional)
    outcsv.writerow(x[0] for x in result.description)

    # dump rows
    outcsv.writerows(result.fetchall())
    outfile.close()

但是，我收到以下错误消息 - 回溯（最近一次通话最后）： 文件“”，第 12 行，在 AttributeError: 'ResultProxy' 对象没有属性 'description'

如果我评论以下命令，我可以成功获得查询结果，但没有标题。 outcsv.writerow(x[0] for x in result.description)

经过研究，我发现了方法——result._metadata.keys 可以生成标题。但是，它会生成某种形式的列表，我无法将其附加为标题。

请告知是否有任何方法可以将文件头以及数据放入 csv 文件中。在回答上述问题时，请考虑我是 Python 初学者这一事实。

Answer 1

在您的示例中的这一行中：

result = engine.execute("""select * from db.tablename """)

变量result 指向sqlalchemy.engine.ResultProxy 类的一个实例。

您希望将列名写入 csv 文件的第一行，并通过检查您的 result 对象找到 result._metadata.keys，该对象返回列名的常规 python list。

Python 中的一个约定是，只要对象上的属性、变量或方法以下划线开头（例如_metadata），这意味着它不打算成为公共 API，并且包的维护者可能会更改他们的如果您依赖这些东西（但对于像 SQLAlchemy 这样的稳定​​库来说不太可能），那么这种方式会破坏您的代码。幸运的是，在这种情况下，有一个记录在案的公共 API 供您获取所需内容：ResultProxy.keys()。

请记住，您的 result 变量指向一个 ResultProxy 实例，因此您可以访问其上的 keys() 方法来获取列名，例如：

result.keys()

当我们尝试写入您的 csv 文件时会发生什么：

outcsv.writerow(result.keys())

我们得到这个异常：

TypeError: a bytes-like object is required, not 'str'

我假设这就是你所说的：

但是，它会生成某种形式的列表，我无法将其附加为 标题。

您可以查看open() API，但问题是您打开文件的方式是它需要二进制数据，而不是从result.keys() 返回的字符串。因此，如果我们将您打开文件的行更改为：

outfile = open('/path/filename.csv', 'w')

outfile.writerow 将接受result.keys() 的结果。

其他一切都应该“正常工作”。

这是我用来测试的代码：

import csv

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)
Session = sessionmaker(bind=engine)

class Model(Base):
    __tablename__ = 'model'
    id = Column(Integer, primary_key=True)
    col1 = Column(String(10))
    col2 = Column(String(10))

if __name__ == '__main__':
    # create some test data
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    s = Session()
    data = dict(col1='a', col2='b')
    s.add_all(Model(**data) for _ in range(5))
    s.commit()
    s.close()
    # put the session away and work with the engine

    result = engine.execute("select * from model")

    outfile = open('filename.csv', 'w', newline='')
    outcsv = csv.writer(outfile, delimiter='\t')
    outcsv.writerow(result.keys())
    outcsv.writerows(result.fetchall())

这是 csv 的内容：

id  col1    col2
1   a   b
2   a   b
3   a   b
4   a   b
5   a   b