Merge pull request #59 from drdee/escape-hstore-array

Escape string in hstore value

Author: dmosorast

tap_postgres/sync_strategies/logical_replication.py

@@ -10,6 +10,7 @@
 import tap_postgres.sync_strategies.common as sync_common
 from dateutil.parser import parse
 import psycopg2
+from psycopg2 import sql
 import copy
 from select import select
 from functools import reduce
@@ -68,11 +69,14 @@ def tuples_to_map(accum, t):
     accum[t[0]] = t[1]
     return accum
 
+def create_hstore_elem_query(elem):
+    return sql.SQL("SELECT hstore_to_array({})").format(sql.Literal(elem))
+
 def create_hstore_elem(conn_info, elem):
     with post_db.open_connection(conn_info) as conn:
         with conn.cursor() as cur:
-            sql = """SELECT hstore_to_array('{}')""".format(elem)
-            cur.execute(sql)
+            query = create_hstore_elem_query(elem)
+            cur.execute(query)
             res = cur.fetchone()[0]
             hstore_elem = reduce(tuples_to_map, [res[i:i + 2] for i in range(0, len(res), 2)], {})
             return hstore_elem

tests/test_discovery.py

@@ -365,6 +365,16 @@ def test_catalog(self):
                                   'definitions' : tap_postgres.BASE_RECURSIVE_SCHEMAS},
                                  stream_dict.get('schema'))
 
+    def test_escaping_values(self):
+        key = 'nickname'
+        value = "Dave's Courtyard"
+        elem = '"{}"=>"{}"'.format(key, value)
+
+        with get_test_connection() as conn:
+          with conn.cursor() as cur:
+            query = tap_postgres.sync_strategies.logical_replication.create_hstore_elem_query(elem)
+            self.assertEqual(query.as_string(cur), "SELECT hstore_to_array('\"nickname\"=>\"Dave''s Courtyard\"')")
+
 
 class TestEnumTable(unittest.TestCase):
     maxDiff = None