SQLAlchemy的，Psycopg2和PostgreSQL COPY

Question

看起来Psycopg具有用于执行COPY自定义命令：

psycopg2 COPY using cursor.copy_from() freezes with large inputs

有没有从SQLAlchemy的访问此功能的方法吗？

Answer 1

It doesn't look like it。

您可能不得不使用psycopg2来公开此功能并放弃ORM功能。我想我无法真正看到ORM在这样的操作中的好处，因为它是一个直接的批量插入，并且处理单个对象，而ORM并不会真正起到很大的作用。

Answer 2

接受的答案是正确的，但如果你想不仅仅是EoghanM的评论多了去了在复制一个表出来，以CSV以下工作对我来说...

from sqlalchemy import sessionmaker, create_engine 

eng = create_engine("postgresql://user:pwd@host:5432/db") 
ses = sessionmaker(bind=engine) 

dbcopy_f = open('/tmp/some_table_copy.csv','wb') 

copy_sql = 'COPY some_table TO STDOUT WITH CSV HEADER' 

fake_conn = eng.raw_connection() 
fake_cur = fake_conn.cursor() 
fake_cur.copy_expert(copy_sql, dbcopy_f) 

的sessionmaker是没有必要的，但如果您习惯于在创建引擎和会话的同时使用raw_connection，则需要将它们分开（除非有某种方式通过会话对象访问引擎，但我不知道）。提供给copy_expert的sql字符串也不是唯一的方法，有一个基本的copy_to函数，您可以使用该参数的子集，您可以使用该参数的子集来查询正常的COPY。对我来说，命令的整体性能似乎很快，将约20000行的表格复制出来。

http://initd.org/psycopg/docs/cursor.html#cursor.copy_to http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.Engine.raw_connection

Answer 3

如果您的发动机配置了psycopg2连接字符串（这是默认的，所以无论是"postgresql://..."或"postgresql+psycopg2://..."），您可以创建一个SQL炼金术会话使用

cursor = session.connection().connection.cursor() 

一个psycopg2光标

，你可以用它来执行

cursor.copy_from(...) 

游标将在当前会话的相同事务中处于活动状态。如果发生commit或rollback，则在抛出psycopg2.InterfaceError的情况下进一步使用游标时，必须创建一个新的游标。

Answer 4

您不需要下拉到psycopg2，使用raw_connection或光标。

只是执行的SQL像往常一样，你甚至可以使用绑定参数与text()：

engine.execute(text('''copy some_table from :csv 
         delimiter ',' csv''' 
        ).execution_options(autocommit=True), 
       csv='/tmp/a.csv') 

您可以删除execution_options(autocommit=True)如果this PR将被接受

Answer 5

您可以使用：

def to_sql(engine, df, table, if_exists='fail', sep='\t', encoding='utf8'): 
    # Create Table 
    df[:0].to_sql(table, engine, if_exists=if_exists) 

    # Prepare data 
    output = cStringIO.StringIO() 
    df.to_csv(output, sep=sep, header=False, encoding=encoding) 
    output.seek(0) 

    # Insert data 
    connection = engine.raw_connection() 
    cursor = connection.cursor() 
    cursor.copy_from(output, table, sep=sep, null='') 
    connection.commit() 
    cursor.close() 

我在5秒内插入200000行而不是4分钟

Answer 6

如果你能得到你把所有的引擎，你需要这样做：

engine = create_engine('postgresql+psycopg2://myuser:password@localhost/mydb') 
# or 
engine = session.engine 
# or any other way you know to get to the engine 

现在你可以工作。

# isolate a connection 
connection = engine.connect().connection 

# get the cursor 
cursor = connection.cursor() 

这里有一些模板的COPY语句来cursor.copy_expert()，因为它表示在这里使用，更全面，更灵活的选择比copy_from()或copy_to()：http://initd.org/psycopg/docs/cursor.html#cursor.copy_expert。

# to dump to a file 
dump_to = """ 
COPY mytable 
TO STDOUT 
WITH (
    FORMAT CSV, 
    DELIMITER ',', 
    HEADER 
); 
""" 

# to copy from a file: 
copy_from = """ 
COPY mytable 
FROM STDIN 
WITH (
    FORMAT CSV, 
    DELIMITER ',', 
    HEADER 
); 
""" 

查看一下上述的均值和其他可能会感兴趣的您的具体情况https://www.postgresql.org/docs/current/static/sql-copy.html的选项。

重要说明：指向cursor.copy_expert()文档的链接指示使用STDOUT写出文件，并使用STDIN从文件复制。但是，如果您查看PostgreSQL手册中的语法，则会注意到您还可以指定要在COPY语句中写入或直接写入的文件。不要这样做，如果你不是以root身份运行（谁在开发过程中以root身份运行Python），那么你可能只是在浪费时间？只需执行psycopg2文档中指出的内容并在您的语句中指定STDIN或STDOUT就可以使用cursor.copy_expert() ，它应该没问题。

# running the copy statement 
with open('/path/to/your/data/file.csv') as f: 
    cursor.copy_expert(copy_from, file=f) 

# don't forget to commit the changes. 
connection.commit()