fix(luigi/contrib/postgres.py): replace copy_from with copy_expert

copy_from does not accept 'schema.table' notation since
psycopg2.9. Then a bug arises in Luigi if the table refers a schema
name.

This PR modifies the 'copy' method behavior, by refering copy_expert
instead of copy_from, as suggested in
psycopg/psycopg2#1294.

Fix #3198

luigi/contrib/postgres.py

@@ -356,16 +356,15 @@ def copy(self, cursor, file):
         else:
             raise Exception('columns must consist of column strings or (column string, type string) tuples (was %r ...)' % (self.columns[0],))
 
-        # cursor.copy_from is not available in pg8000
-        if hasattr(cursor, 'copy_from'):
-            cursor.copy_from(
-                file, self.table, null=r'\\N', sep=self.column_separator, columns=column_names)
+        copy_sql = (
+            "COPY {table} ({column_list}) FROM STDIN "
+            "WITH (FORMAT text, NULL '{null_string}', DELIMITER '{delimiter}')"
+        ).format(table=self.table, delimiter=self.column_separator, null_string=r'\\N',
+                 column_list=", ".join(column_names))
+        # cursor.copy_expert is not available in pg8000
+        if hasattr(cursor, 'copy_expert'):
+            cursor.copy_expert(copy_sql, file)
         else:
-            copy_sql = (
-                "COPY {table} ({column_list}) FROM STDIN "
-                "WITH (FORMAT text, NULL '{null_string}', DELIMITER '{delimiter}')"
-            ).format(table=self.table, delimiter=self.column_separator, null_string=r'\\N',
-                     column_list=", ".join(column_names))
             cursor.execute(copy_sql, stream=file)
 
     def run(self):