Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_dynamic_query(query: str, **identifiers: str) -> str:
return psycopg2_sql.SQL(query).format(
**{key: psycopg2_sql.Identifier(value) for key, value in identifiers.items()}
)
def copy_site_preferences(apps, schema_editor):
with con.cursor() as cursor:
table_names = [
citem.name for citem in con.introspection.get_table_list(cursor)]
if 'siteprefs_preference' not in table_names:
return
# Remove left over siteprefs for app email_action
cursor.execute(
sql.SQL(
'DELETE FROM siteprefs_preference WHERE app = {0}').format(
sql.Literal('email_action')))
# Change logs app to ontask
cursor.execute(
sql.SQL('UPDATE siteprefs_preference '
+ 'SET app={0} where app={1}').format(
sql.Literal('ontask'),
sql.Literal('logs')))
# Change max_list_size to max_log_list_size
cursor.execute(
sql.SQL('UPDATE siteprefs_preference '
+ 'SET name={0} where name={1}').format(
sql.Literal('max_log_list_size'),
sql.Literal('max_list_size')))
def _make_dynamic_query(self, query: str, **identifiers: str) -> str:
return psycopg2.sql.SQL(query).format(
**{
key: psycopg2.sql.Identifier(value)
for key, value in identifiers.items()
}
def rename_table(table: str, new_name: str):
"""Rename a table in the database.
:param table: Current table name
:param new_name: New table name
:return: Nothing. Change reflected in the database table
"""
with connection.connection.cursor() as cursor:
cursor.execute(sql.SQL('ALTER TABLE {0} RENAME TO {1}').format(
sql.Identifier(table),
sql.Identifier(new_name),
))