python之SQLAlchemy之如何从 ResultProxy 访问列名并写入 CSV header_编程设计

我正在尝试使用 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 初学者这一事实。

请您参考如下方法:

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

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