python - SQLAlchemy - 如何从 ResultProxy 访问列名并写入 CSV header

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)，这意味着它不打算成为公共(public) API，并且维护者如果您依赖这些东西，该包可能会以破坏您的代码的方式更改它们的实现(但是对于像 SQLAlchemy 这样的稳定​​库来说不太可能)。幸运的是，在这种情况下，有一个记录在案的公共(public) API 供您获取所需内容:ResultProxy.keys() .

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

result.keys()

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

outcsv.writerow(result.keys())

我们得到这个异常:

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

我假设这就是您所说的:

However, it generates in some form of a list that I can't attach as header.

你可以看看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