是否可以通过psycopg2运行下面的命令？ 如果可以，我怎样才能做到？

COPY table_name(col1,col2) FROM 'path/to/file.csv' WITH HEADER DELIMITER ',' CSV;
    
python
postgresql
psycopg2
user680839
发布于 2013-07-10
4 个回答
Sean
发布于 2020-07-06
已采纳
0 人赞同

你可以使用copy_from method:
import psycopg2
dbname=...
user=...
password=...
host=...
port=...
con = psycopg2.connect(database=dbname,user=user,password=password,host=host,port=port)
cur = con.cursor()    
f = open('path/to/file.csv')
cur.copy_from(f, 'test', columns=('col1', 'col2'), sep=",")
con.commit()
con.close()
    
shrinathM：
我们能否提到copy_from方法的引号。我在csv文件中的字符串列周围有双引号？
Brad Koch：
@shrinathM psycopg2的copy_from目前不支持带引号的值，当你有一个非琐碎的CSV文件时，请使用copy_expert代替。
Craig Ringer
发布于 2020-07-06
0 人赞同

在谷歌上搜索psycopg2 copy，发现，我的第一个点击。心理咨询师手册, which includes instructions on using client-side COPY.
如果你想要服务器端的COPY，你只需像其他SQL一样运行该语句。
正如上面写的那样，这个命令没有任何意义。我推测你是想写一个COPY ... TO或COPY ... FROM的命令，并在隐藏真正的文件名等情况下把它弄乱了。
    
user680839：
你是对的，我的命令中有一个错字。我忘了运行cur.commit()。谢谢你的帮助!
Brad Koch
发布于 2020-07-06
0 人赞同

The copy_from命令对基本的使用情况很有用。
with open('path/to/file.csv') as f:
    cursor.copy_from(f, 'table_name', columns=('col1', 'col2'), sep=',')
请注意，当按照你的问题中描述的那样复制CSV数据时，各种问题会阻止你使用copy_from--头行、带引号的值、包含逗号的值。  CSVs可以使用copy_expert语句，用一个简单的手动组成的COPY查询。
with open('path/to/file.csv') as f:
    cursor.copy_expert('COPY table_name(col1, col2) FROM STDIN WITH HEADER CSV', f)
    
hansaplast
发布于 2020-07-06
0 人赞同

使用copy_from有一个问题。它不能识别带引号的字段，例如，如果你有一个值with, a comma，并使用csv.writer，那么它就被写成了,"with, a comma"。 psycopg2不能识别这一点（见@shrinathM的quotestring评论）。
这就是为什么在大多数情况下，你需要退回到更基本的copy_expert。
也就是说，在使用COPY时，我发现的另一个困难是要理解CSV必须如何结构化，以使postgres正确接收它。这里有一个使用StringIO而不是文件的基本代码。
下面的解决方案告诉你如何用python编写CSV，以及如何使用copy_expert。如果你已经有了CSV，那么你只需要最后一行。
import io
import csv
import datetime 
f = io.StringIO()
w = csv.writer(f)
data = [
  ['Hans', [1,2,3], True],
  ['Kurt', [4], False],
columns = ['name', 'ids', 'has_foo']
print('convert to csv format')
for l_in in data:
  l_out = []
  for v in l_in:
    if v == None:
      l_out.append('')
    elif type(v) in [str, int, datetime.date]:
      l_out.append(str(v))
    elif type(v) in [list, set, tuple]:
      l_out.append('{' + ','.join(str(i) for i in v) + '}')
    elif type(v) == bool:
      if v:
        l_out.append('t')
      else:
        l_out.append('f')
    else:
      print(f'unsupported type {type(v)}, writing str()')
      l_out.append(str(v))
  w.writerow(l_out)
print('actual copy')