psycopg2.sql how to write a "CREATE TABLE..." statement specifying column names + types?

[henryli78]

I'm trying to convert older code that directly ran psycopg2 commands via cursor.execute(sql) where sql was a string to use psycopg2.sql instead.

I was able to modify most statements I had but had trouble with this:

Old Python code:

schema="public"
table_name="test"
col_types = [
    ("a", "integer"),
    ("b", "text"),
]

...
with conn.cursor() as cur:
  sql = f"CREATE TABLE IF NOT EXISTS {schema}.{table_name}"
  columns = ", ".join([f'"{name}" {type}' for name, type in col_types])
  
  sql += f" ({columns})"
  try:
      cur.execute(sql)
      conn.commit()
      logging.info(f"create table: {sql=}")
  except (Exception, psycopg2.DatabaseError) as error:
      logging.error(error)
      conn.rollback()

Logged output:

INFO  || create table: CREATE TABLE IF NOT EXISTS public.test ("a" integer, "b" text)

Attempted new code:

from psycopg2.sql import SQL, Identifier, Placeholder

schema="public"
table_name="test"
col_types = [
    ("a", "integer"),
    ("b", "text"),
]
...
with conn.cursor() as cur:
  sql = SQL("CREATE TABLE IF NOT EXISTS {name} ({cols})").format(
      name=Identifier(schema, table_name),
      cols=SQL(", ").join(
          SQL(" ").join([Identifier(name), SQL(type)])for name, type in self.col_types
      ),
  )

  try:
      cur.execute(sql)
      conn.commit()
      logging.info("create table: %s", sql)
  except (Exception, psycopg2.DatabaseError) as error:
      logging.exception(error)

Logged output:

Traceback (most recent call last):
  File "/home/henryli78/database_test.py", line 54, in <module>
    cur.execute(sql)
psycopg2.errors.SyntaxError: syntax error at or near ""a""
LINE 1: ...LE IF NOT EXISTS "public"."test" "a" intege...

Does anyone know what the correct syntax is for creating a table with psycopg2 using psycopg2.sql and be able to specific column names/types?

[dvarrazzo]

The error message is not consistent with your code: you have brackets in you code ({cols}) that don't appear in the message.

You can use print(sql.as_string(conn)) to figure out what you are doing wrong.

[henryli78]

You're right. I made a typo locally -- thanks for pointing out my mistake. print(sql.as_string(conn)) was a helpful tool as well.