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”对象没有属性“描述”

如果我评论下面的命令，我可以成功地获得查询结果但没有标题。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，该对象返回list列名的常规 python 。

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