X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/c7f40e3cee2853ade0e16a90ab7a3b29b02265f9..4f59644cc2ef6da59f96bee2efdd722d943677b1:/nominatim/db/utils.py?ds=inline

diff --git a/nominatim/db/utils.py b/nominatim/db/utils.py
index 6d2eb297..87b0545f 100644
--- a/nominatim/db/utils.py
+++ b/nominatim/db/utils.py
@@ -1,12 +1,19 @@
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
 """
 Helper functions for handling DB accesses.
 """
 import subprocess
 import logging
 import gzip
+import io
 
-from .connection import get_pg_env
-from ..errors import UsageError
+from nominatim.db.connection import get_pg_env
+from nominatim.errors import UsageError
 
 LOG = logging.getLogger()
 
@@ -35,24 +42,72 @@ def execute_file(dsn, fname, ignore_errors=False, pre_code=None, post_code=None)
         cmd.append('--quiet')
     proc = subprocess.Popen(cmd, env=get_pg_env(dsn), stdin=subprocess.PIPE)
 
-    if not LOG.isEnabledFor(logging.INFO):
-        proc.stdin.write('set client_min_messages to WARNING;'.encode('utf-8'))
-
-    if pre_code:
-        proc.stdin.write((pre_code + ';').encode('utf-8'))
+    try:
+        if not LOG.isEnabledFor(logging.INFO):
+            proc.stdin.write('set client_min_messages to WARNING;'.encode('utf-8'))
 
-    if fname.suffix == '.gz':
-        with gzip.open(str(fname), 'rb') as fdesc:
-            remain = _pipe_to_proc(proc, fdesc)
-    else:
-        with fname.open('rb') as fdesc:
-            remain = _pipe_to_proc(proc, fdesc)
+        if pre_code:
+            proc.stdin.write((pre_code + ';').encode('utf-8'))
 
-    if remain == 0 and post_code:
-        proc.stdin.write((';' + post_code).encode('utf-8'))
+        if fname.suffix == '.gz':
+            with gzip.open(str(fname), 'rb') as fdesc:
+                remain = _pipe_to_proc(proc, fdesc)
+        else:
+            with fname.open('rb') as fdesc:
+                remain = _pipe_to_proc(proc, fdesc)
 
-    proc.stdin.close()
+        if remain == 0 and post_code:
+            proc.stdin.write((';' + post_code).encode('utf-8'))
+    finally:
+        proc.stdin.close()
+        ret = proc.wait()
 
-    ret = proc.wait()
     if ret != 0 or remain > 0:
         raise UsageError("Failed to execute SQL file.")
+
+
+# List of characters that need to be quoted for the copy command.
+_SQL_TRANSLATION = {ord(u'\\'): u'\\\\',
+                    ord(u'\t'): u'\\t',
+                    ord(u'\n'): u'\\n'}
+
+
+class CopyBuffer:
+    """ Data collector for the copy_from command.
+    """
+
+    def __init__(self):
+        self.buffer = io.StringIO()
+
+
+    def __enter__(self):
+        return self
+
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        if self.buffer is not None:
+            self.buffer.close()
+
+
+    def add(self, *data):
+        """ Add another row of data to the copy buffer.
+        """
+        first = True
+        for column in data:
+            if first:
+                first = False
+            else:
+                self.buffer.write('\t')
+            if column is None:
+                self.buffer.write('\\N')
+            else:
+                self.buffer.write(str(column).translate(_SQL_TRANSLATION))
+        self.buffer.write('\n')
+
+
+    def copy_out(self, cur, table, columns=None):
+        """ Copy all collected data into the given table.
+        """
+        if self.buffer.tell() > 0:
+            self.buffer.seek(0)
+            cur.copy_from(self.buffer, table, columns=columns)