sqltools: avoid copy_from, psycopg 2.9 broke it

Author: markokr

skytools/basetypes.py

@@ -38,8 +38,7 @@ def fetchall(self) -> Sequence[DictRow]: raise NotImplementedError
     def fetchone(self) -> DictRow: raise NotImplementedError
     def __enter__(self) -> "Cursor": raise NotImplementedError
     def __exit__(self, typ: Type, value: Any, traceback: Any) -> Any: raise NotImplementedError
-    def copy_from(self, buf: IO[str], hdr: str) -> None: raise NotImplementedError
-    def copy_expert(self, sql: str, f: Union[IO[str], io.TextIOBase]) -> None: raise NotImplementedError
+    def copy_expert(self, sql: str, f: Union[IO[str], io.TextIOBase], size: int = 8192) -> None: raise NotImplementedError
 
 
 class Connection(Protocol):

skytools/sqltools.py

@@ -341,8 +341,8 @@ def magic_insert(curs: Cursor, tablename: str,
         curs.execute(buf.getvalue())
     else:
         buf.seek(0)
-        hdr = "%s (%s)" % (qtablename, ",".join(qfields))
-        curs.copy_from(buf, hdr)
+        sql = "COPY %s (%s) FROM STDIN" % (qtablename, ",".join(qfields))
+        curs.copy_expert(sql, buf)
     return None
 
 
@@ -401,9 +401,10 @@ def flush(self) -> None:
 
         self.buf.seek(0)
         if self.sql_from:
-            self.dstcurs.copy_expert(self.sql_from, self.buf)
+            sql = self.sql_from
         else:
-            self.dstcurs.copy_from(self.buf, self.tablename or "missing_table_name")
+            sql = "COPY %s FROM STDIN" % (self.tablename or "missing_table_name",)
+        self.dstcurs.copy_expert(sql, self.buf)
         self.buf.seek(0)
         self.buf.truncate()