Update selected rows of a column in table using python

This recipe lets you to perform slightly more complex update operations. With more power comes more challenges. We will explore them in this notebook. If you wish to have finer control consider writing custom SQL queries in raw query recipe.

This notebook was created with MLJAR Studio

MLJAR Studio is Python code editior with interactive code recipes and local AI assistant.
You have code recipes UI displayed at the top of code cells.

Documentation

All required packages are automatically imported by MLJAR Studio for you so you don't have to worry about them.

# import packages
import psycopg
import os
from dotenv import load_dotenv
from psycopg import sql

Make sure you opened a connection in your notebook. To learn how to do it, check out open and test database connection example notebook.

# load credentials from .env file:
load_dotenv(override=True)

# get the credentials
def create_new_connection():
    try:
        conn = psycopg.connect(
            dbname=os.getenv("POSTGRES_DB_NAME"),
            user=os.getenv("POSTGRES_USERNAME"),
            password=os.getenv("POSTGRES_PASSWORD"),
            host=os.getenv("POSTGRES_HOST"),
            port=os.getenv("POSTGRES_PORT"),
        )
        return conn
    # check for errors
    except psycopg.Error as e:
        raise psycopg.Error(f"""
Error occurred while establishing connection: 
    {e}

Maybe you provided wrong credentials, use define new connection recipe to edit them.
Other option is that database server is down or you dont have the right acces to this database.
            """)

# open new connection:
conn = create_new_connection()

We will perform our updates in products table. Lets list all columns of that table using show all columns recipe

# if connection was used and closed it is reopen here
if conn.closed:
    conn = create_new_connection()

# run query
with conn:
    with conn.cursor() as cur:

        # query db
        try:
            cur.execute("""
                    SELECT attname AS col, atttypid::regtype AS datatype
                    FROM pg_attribute
                    WHERE attrelid = %s::regclass 
                    AND attnum > 0
                    AND NOT attisdropped
                    ORDER BY attnum;""", 
                    ("products",))
        # check for errors
        except psycopg.ProgrammingError as e:
            raise psycopg.ProgrammingError(f"""
Problem running query:
    {e}

Did you spell everything correctly?
You can use show tables and columns recipes.
            """)

        # print the results
        print("Columns of products:")
        for column in cur.fetchall():
            print(f"{column}")

Lets check the contents of products table using select query recipe.

# if connection was used and closed it is reopen here
if conn.closed:
    conn = create_new_connection()

# run query
with conn:
    with conn.cursor() as cur:

        # query db
        try:
            cur.execute(
                sql.SQL("SELECT {columns} FROM {table}").format(
                    columns=sql.SQL("*"),
                    table=sql.Identifier("products"),
                )
            )
        # check for errors
        except psycopg.ProgrammingError as e:
            raise psycopg.ProgrammingError(f"""
Problem running query:
    {e}

Did you spell everything correctly?
You can use show tables and columns recipes.
            """)

        # print the results
        for row in cur.fetchall():
            print(f"{row}")

We can see that the description column is empty. Lets fill it out. To do so we will use Update selected column recipe.

Lets fill out our form. Set the table and column name and lets check show results. This will display updated rows under the code field. If we set id field to contain only one number it will update it, and all subsequent rows until it runs out of data to insert.

# if connection was used and closed it is reopen here
if conn.closed: 
    conn = create_new_connection()

# run query
with conn:
    with conn.cursor() as cur:

        # update data
        try: 
            query = sql.SQL(
                "UPDATE {table} SET {column} = %(value)s WHERE id = %(id)s RETURNING *"
            ).format(table=sql.Identifier("products"), column=sql.Identifier("description"))
            cur.executemany(query, params_seq=[{"id": 1, "value": "first description"}, {"id":  2, "value": " second description"}, {"id":  3, "value": " third description"}], returning=True)
        # check for errors
        except psycopg.ProgrammingError as e:
            raise psycopg.ProgrammingError(f"""
Problem running query:
    {e}

Are you sure every name is spelled correctly?
You can use show all tables or columns to check db contents.
            """)

        # print the results
        while True:
            for row in cur.fetchall():
                print(row)
            if not cur.nextset():
                break

Another way to achieve the same result is to use insert from list option. We'll need to make a python list of ids and vales to later pass it into the recipe.

# we will update on 2,4,5 ids
ids = [2,4,5]
# those are the values to be inserted
descriptions = ["second description", "forth description", "fifth description"]

Lets use update selected columns recipe again, but this time tic the update from list toggle. Notice that id and value fields changed and now let us to select form all the lists in our notebook. Lets select ids and descriptions list respectively.

# if connection was used and closed it is reopen here
if conn.closed: 
    conn = create_new_connection()

# run query
with conn:
    with conn.cursor() as cur:

        # update data
        try: 
            query = sql.SQL(
                "UPDATE {table} SET {column} = %(value)s WHERE id = %(id)s RETURNING *"
            ).format(table=sql.Identifier("products"), column=sql.Identifier("description"))
            cur.executemany(query, params_seq=[{"id": ids[i], "value": descriptions[i]} for i in range(len(ids))], returning=True)
        # check for errors
        except psycopg.ProgrammingError as e:
            raise psycopg.ProgrammingError(f"""
Problem running query:
    {e}

Are you sure every name is spelled correctly?
You can use show all tables or columns to check db contents.
            """)

        # print the results
        while True:
            for row in cur.fetchall():
                print(row)
            if not cur.nextset():
                break

As a last thing lest review our changes using select query like in the beginning of this notebook.

# if connection was used and closed it is reopen here
if conn.closed:
    conn = create_new_connection()

# run query
with conn:
    with conn.cursor() as cur:

        # query db
        try:
            cur.execute(
                sql.SQL("SELECT {columns} FROM {table}").format(
                    columns=sql.SQL("*"),
                    table=sql.Identifier("products"),
                )
            )
        # check for errors
        except psycopg.ProgrammingError as e:
            raise psycopg.ProgrammingError(f"""
Problem running query:
    {e}

Did you spell everything correctly?
You can use show tables and columns recipes.
            """)

        # print the results
        for row in cur.fetchall():
            print(f"{row}")

Conclusions

Updating tables might be simple or very complicated depending on your use case. With this recipe those simple and some more advanced cases will become almost trivial.

Recipes used in the postgresql-python-update-selected.ipynb

All code recipes used in this notebook are listed below. You can click them to check their documentation.

Packages used in the postgresql-python-update-selected.ipynb

List of packages that need to be installed in your Python environment to run this notebook. Please note that MLJAR Studio automatically installs and imports required modules for you.

psycopg>=3.2.1

python-dotenv>=1.0.1

Similar notebooks

List of similar Python notebooks, so you can find more inspiration 😊