79543682

Date: 2025-03-29 18:10:09
Score: 0.5
Natty:
Report link

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:

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()
Reasons:
  • Blacklisted phrase (1): this blog
  • Long answer (-1):
  • Has code block (-0.5):
  • Low reputation (1):
Posted by: Matt