I know it sounds silly and perhaps counter-intuitive but let us say that you are writing a Python script that does some logic to that eventually has to update the database table based on external data sources. A flip here and there and you find that you have to write yourself an UPDATE SQL statement and execute it on the database side.

Fabric can only do so much

I use fabric to help execute remotely. If it's something repetitive, it is best to try to automate the task.

from fabric import task
from getpass import getpass
from invoke import Responder

db_pass = getpass("Enter DB password: ")
db_responder = Responder(
    pattern = r"Password:",
    response = f"{db_pass}\n"
)

@task
def update_rows(connection):
    result = connection.run('su postgres -c "psql -d customer -c "UPDATE customers SET verified=False WHERE id IN (\'123\', \'456\')";"', pty=True, watchers=[db_pass])

The problem with the fabric script above are 2 things:

1. The juggle between single quotes and double quotes is a nightmare to write and handle. And even if it was successfully escaped, bash or database will scream back that there's an error or something.

2. Passing a list in Python is, by default, square brackets [ ]. Easy fix by moving around and formatting into a .join() string but, back to the first problem again, they need to be enclosed in single quoted strings.

After a few tries, I've tested that passing a SQL statement to the psql client via SQL file works but then that means I have to generate the SQL statements first.

Enter SQLAlchemy

I don't need to connect to the database. I already have the database table schema and I know what to update. I just need an easy way to generate SQL statements without the need to join and juggling with single quotes and double quotes. pip install sqlalchemy and off I go.

from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, mapped_column
from sqlalchemy import update

class Base(DeclarativeBase):
    pass

class Customers(Base):
    __tablename__ = 'customers'
    name = mapped_column(postgresql.VARCHAR)
    id = mapped_column(postgresql.BIGINT, primary_key=True)
    verified = mapped_column(postgresql.BOOLEAN)
    address = mapped_column(postgresql.VARCHAR)
    phone = mapped_column(postgresql.VARCHAR)


    def __str__(self):
        pass


def reset_customer_status(customer_ids):
    q = update(Customers).where(Customers.id.in_(customer_ids)).values(verified=0)
    return q.compile(dialect=postgresql.dialect(), compile_kwargs={'literal_binds': True})


if __name__ == '__main__':
    ids_to_reset = [
        'ABC1234DEF',
        'ABC1111DEF',
        'ABC2222DEF',
    ]
    print(reset_customer_status(ids_to_reset))

The output of running the above script shows:

UPDATE customers SET verified=false WHERE customers.id IN ('ABC1234DEF', 'ABC1111DEF', 'ABC2222DEF')

That is nearly perfect! Now I can easily pass in a list of IDs and then it will nearly handle the rest of the conversion and execution by SQL statements into a SQL file. 🥳

with open('update_reset_customer_status.sql', 'w') as f:
    f.write(reset_customer_status(ids_to_reset))

Then I have to rewrite my fabric function to put the file in the database server.

@task
def upload_sqlfile(connection, sqlfile):
    connection.put(sqlfile, '~/')

Then in my terminal, fab -H database.pr.local upload_sqlfile update_reset_customerstatus.sql --prompt-for-login-password

Honestly this took a whole day but I can pretty much guarantee that this is going to be a frequent request, so having this script is going to be pretty handy.

I could probably turn this into a web app (Flask or Django maybe?) and with a few clicks and taps, I can have it running in the browser. That's probably for another post at another time.