bn8/python-mylib
1
0
Fork 0
Code Issues Pull requests Projects Releases 12 Wiki Activity Actions

PgSQL: fix quoting table name with schema prefix

Browse source
This commit is contained in:
Benjamin Renard 2021-11-07 21:57:00 +01:00
parent 6b825813c1
commit c77f9e5674
2 changed files with 21 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
mylib/pgsql.py
View file

@ -254,10 +254,19 @@ class PgDB:
sql += " WHERE " + sql_where sql += " WHERE " + sql_where
return (sql, params) return (sql, params)
@staticmethod
def _quote_table_name(table):
""" Quote table name """
return '"{0}"'.format(
'"."'.join(
table.split('.')
)
)
def insert(self, table, values, just_try=False): def insert(self, table, values, just_try=False):
""" Run INSERT SQL query """ """ Run INSERT SQL query """
sql = 'INSERT INTO "{0}" ("{1}") VALUES ({2})'.format( sql = 'INSERT INTO {0} ("{1}") VALUES ({2})'.format(
table, self._quote_table_name(table),
'", "'.join(values.keys()), '", "'.join(values.keys()),
", ".join([ ", ".join([
'%({0})s'.format(key) '%({0})s'.format(key)
@ -277,8 +286,8 @@ class PgDB:
def update(self, table, values, where_clauses, where_op=None, just_try=False): def update(self, table, values, where_clauses, where_op=None, just_try=False):
""" Run UPDATE SQL query """ """ Run UPDATE SQL query """
sql = 'UPDATE "{0}" SET {1}'.format( sql = 'UPDATE {0} SET {1}'.format(
table, self._quote_table_name(table),
", ".join([ ", ".join([
'"{0}" = %({0})s'.format(key) '"{0}" = %({0})s'.format(key)
for key in values for key in values
@ -304,7 +313,7 @@ class PgDB:
def delete(self, table, where_clauses, where_op='AND', just_try=False): def delete(self, table, where_clauses, where_op='AND', just_try=False):
""" Run DELETE SQL query """ """ Run DELETE SQL query """
sql = 'DELETE FROM "{0}"'.format(table) sql = 'DELETE FROM {0}'.format(self._quote_table_name(table))
params = dict() params = dict()
try: try:
@ -326,7 +335,7 @@ class PgDB:
def truncate(self, table, just_try=False): def truncate(self, table, just_try=False):
""" Run TRUNCATE SQL query """ """ Run TRUNCATE SQL query """
sql = 'TRUNCATE "{0}"'.format(table) sql = 'TRUNCATE {0}'.format(self._quote_table_name(table))
if just_try: if just_try:
log.debug("Just-try mode: execute TRUNCATE query: %s", sql) log.debug("Just-try mode: execute TRUNCATE query: %s", sql)
@ -348,7 +357,7 @@ class PgDB:
else: else:
sql += '"{0}"'.format('", "'.join(fields)) sql += '"{0}"'.format('", "'.join(fields))
sql += ' FROM "{0}"'.format(table) sql += ' FROM {0}'.format(self._quote_table_name(table))
params = dict() params = dict()
try: try:

5
tests/test_pgsql.py
View file

@ -251,6 +251,11 @@ def test_add_where_clauses_with_duplicated_field():
) )
def test_quote_table_name():
assert PgDB._quote_table_name("mytable") == '"mytable"'
assert PgDB._quote_table_name("myschema.mytable") == '"myschema"."mytable"'
def test_insert(mocker, test_pgdb): def test_insert(mocker, test_pgdb):
values = dict(test1=1, test2=2) values = dict(test1=1, test2=2)
mocker.patch( mocker.patch(