I know this is a very old question, but maybe this will help someone... I had to insert columns into PostgreSQL tables at specific positions many times, so I wrote a Python script for it. It follows the column rotation approach in this blog post, which was also referred to earlier.
Limitations include:
Foreign keys, constraints, indexes etc. will need to be recreated if they apply to columns behind the new one
It is tested with psycopg 3.2.5 and Python 3.13. Other versions may well work, but you would have to try it out
DB connection and script parameters are hardcoded (but then, this is not something to be widely disseminated)
It does handle all data types including arrays and default values. The parameters are hopefully self-explanatory. The new column will be at position new_column_pos
after the script ran.
from psycopg import sql, connect
def insert_pg_column(conn, table_name, new_column_name, new_column_type, new_column_position):
cur = conn.cursor()
# Get column names and types from the table
cur.execute(
sql.SQL(
"SELECT column_name, data_type, udt_name, character_maximum_length, column_default "
"FROM information_schema.columns WHERE table_name = %s ORDER BY ordinal_position"
),
[table_name],
)
columns = cur.fetchall()
print(f"Retrieved definitions for {len(columns)} columns")
# Remove from list all columns which remain unchanged
columns = columns[new_column_position - 1 :]
column_names = [col[0] for col in columns]
# Add the new column to the table (at the end)
cur.execute(
sql.SQL("ALTER TABLE {} ADD COLUMN {} {}").format(
sql.Identifier(table_name), sql.Identifier(new_column_name), sql.SQL(new_column_type)
)
)
print(f"Added new column '{new_column_name}' to table '{table_name}'")
# Create temporary columns to hold the data temporarily
temp_columns = {}
for col_name, col_type, udt_name, length, default in columns:
temp_col_name = f"{col_name}_temp"
temp_columns[col_name] = temp_col_name
# Handle array types
if col_type == "ARRAY":
if udt_name.startswith("_"):
data_type = f"{udt_name[1:]}[]" # Remove the leading underscore
else:
data_type = f"{udt_name}[]" # Not sure this ever happens?
else:
data_type = col_type
if length is not None: # For character types
data_type += f"({length})"
cur.execute(
sql.SQL("ALTER TABLE {} ADD COLUMN {} {} {}").format(
sql.Identifier(table_name),
sql.Identifier(temp_col_name),
sql.SQL(data_type),
sql.SQL("DEFAULT {}").format(sql.SQL(default)) if default is not None else sql.SQL(""),
)
)
print(f"Added temporary column '{temp_col_name}'{(" with default '" + default) + "'" if default else ''}")
# Update the temporary columns to hold the data in the desired order
for col_name in column_names:
cur.execute(
sql.SQL("UPDATE {} SET {} = {}").format(
sql.Identifier(table_name), sql.Identifier(temp_columns[col_name]), sql.Identifier(col_name)
)
)
print(f"Copied data from column '{col_name}' to '{temp_columns[col_name]}'")
# Drop the original columns
for col_name in column_names:
cur.execute(
sql.SQL("ALTER TABLE {} DROP COLUMN {}").format(sql.Identifier(table_name), sql.Identifier(col_name))
)
print(f"Dropped original column '{col_name}'")
# Rename the temporary columns to the original column names
for col_name in column_names:
cur.execute(
sql.SQL("ALTER TABLE {} RENAME COLUMN {} TO {}").format(
sql.Identifier(table_name), sql.Identifier(temp_columns[col_name]), sql.Identifier(col_name)
)
)
print(f"Renamed '{temp_columns[col_name]}' to '{col_name}'")
conn.commit()
cur.close()
if __name__ == "__main__":
# Database connection parameters
HOST = "your_host"
DATABASE = "your_dbname"
USER = "your_user" # Needs to have sufficient privileges to alter the table!
PASSWORD = "your_password"
# Parameters for adding a new column (EXAMPLE; REPLACE WITH YOUR OWN VALUES!)
table_name = "users"
new_column_name = "user_uuid"
new_column_type = "uuid"
new_column_pos = 3 # Position is 1-based index
connection = connect(f"dbname={DATABASE} user={USER} password={PASSWORD} host={HOST}")
try:
insert_pg_column(connection, table_name, new_column_name, new_column_type, new_column_pos)
print(f"Successfully added column '{new_column_name}' to table '{table_name}' at position {new_column_pos}.")
except Exception as e:
print(f"Error: {e}")
connection.rollback()
finally:
connection.close()