Create Table From Dictionary Data In A Safe Way

I have a problem where i have a list of dictionaries with for example the following data: columns = [{ 'name': 'column1', 'type': 'varchar' }, { 'name': 'column2',

Solution 1:

You can make use of AsIs to get the column types added non-quoted:

import psycopg2
from psycopg2.extensions import AsIs
import psycopg2.sql as sql

conn = psycopg2.connect("dbname=mf port=5959 host=localhost user=mf_usr")

columns = [{
    'name': "column1",
    'type': "varchar"
},
{
    'name': "column2",
    'type': "decimal"
}]

# create a dict, so we can use dict placeholders in the CREATE TABLE query.
column_dict = {c['name']: AsIs(c['type']) for c in columns}

createSQL = sql.SQL("CREATE TABLE some_table_name\n ({columns})").format(
    columns = sql.SQL(',').join(
        sql.SQL(' ').join([sql.Identifier(col), sql.Placeholder(col)]) for col in column_dict)
)

print(createSQL.as_string(conn))
cur = conn.cursor()
cur.execute(createSQL, column_dict)
cur.execute("insert into some_table_name (column1) VALUES ('foo')")
cur.execute("select * FROM some_table_name")
print('Result: ', cur.fetchall())

Output:

CREATETABLE some_table_name
 ("column1" %(column1)s,"column2" %(column2)s)
Result:  [('foo', None)]

Note: psycopg2.sql is safe to SQL injection, AsIs probably not. Testing using 'type': "varchar; DROP TABLE foo" resulted in Postgres syntax error:

b'CREATE TABLE some_table_name\n ("column1" varchar; DROP TABLE foo,"column2" decimal)'
Traceback (most recent calllast):
  File "pct.py", line 28, in<module>
    cur.execute(createSQL, column_dict)
psycopg2.errors.SyntaxError: syntax error ator near ";"
LINE 2:  ("column1" varchar; DROPTABLE foo,"column2" decimal)

Solution 2:

Expanding on my comment, a complete example:

import psycopg2 
from psycopg2 import sql

columns = [{
    'name': 'column1',
    'type': 'varchar'
},
{
    'name': 'column2',
    'type': 'decimal'
}]

con = psycopg2.connect("dbname=test host=localhost user=aklaver") 
cur = con.cursor()

col_list = sql.SQL(',').join( [sql.Identifier(col["name"]) + sql.SQL(' ') + sql.SQL(col["type"]) for col in columns])
create_sql = sql.SQL("CREATE TABLE tablename ({})").format(col_list)

print(create_sql.as_string(con))                                                                                                                                          
CREATE TABLE tablename ("column1" varchar,"column2"decimal)

cur.execute(create_sql)
con.commit()


test(5432)=> \d tablename 
                   Table "public.tablename"
 Column  |       Type        | Collation | Nullable | Default 
---------+-------------------+-----------+----------+---------
 column1 | character varying |           |          | 
 column2 | numeric           | 

Iterate over the column list of dicts and assign the column names as SQL identifiers and the column types as straight SQL into sql.SQL construct. Use this as parameter to CREATE TABLESQL.

Caveat: sql.SQL() does not do escaping, so those values would have to be validated before they where used.