4 Handy Python Scripts for Creating & Updating PostgreSQL Databases

Hi there! I'm Shrijith Venkatrama, the founder of Hexmos. Right now, I’m building LiveAPI, a super-convenient tool that simplifies engineering workflows by generating awesome API docs from your code in minutes.

Every now and then - it so happens that I need a new PostgreSQL database to do some quick demo or experiment.

Usually, I have 4 pieces of information with me about a PostgreSQL instance ("the four"):

1. User
2. Password
3. Host
4. Port

Q1: How do I create a new postgresql database given "the four"?

Install Astral's uv from here

Create a new project folder:

mkdir sample_db

Initialize python project:

uv init
uv add psycopg2-binary # postgresql driver

Create a new file create_db.py:

import psycopg2
from psycopg2 import sql

# Database connection details
host = "your_host"
port = "your_port"
user = "your_user"
password = "your_password"
dbname = "postgres"  # Default database to connect to

# Database to create
new_dbname = "my_sample_db"

try:
    # Connect to the PostgreSQL instance
    conn = psycopg2.connect(
        host=host,
        port=port,
        user=user,
        password=password,
        dbname=dbname
    )
    conn.autocommit = True  # Enable autocommit for database creation
    cursor = conn.cursor()

    # Create the new database
    create_db_query = sql.SQL("CREATE DATABASE {}").format(sql.Identifier(new_dbname))
    cursor.execute(create_db_query)
    print(f"Database '{new_dbname}' created successfully.")

    # Verify the database was created
    cursor.execute("SELECT datname FROM pg_database;")
    databases = cursor.fetchall()
    if (new_dbname,) in databases:
        print(f"Database '{new_dbname}' exists in the PostgreSQL instance.")
    else:
        print(f"Database '{new_dbname}' does not exist in the PostgreSQL instance.")

except psycopg2.Error as e:
    print(f"An error occurred: {e}")
finally:
    # Close the connection
    if conn:
        cursor.close()
        conn.close()
        print("Connection closed.")

Run with:

uv run python create_db.py

Q2: How to create a "Users" table in "my_sample_db"?

Since the "connection" part is common to all our future tasks, let's create a common file database_connection.py:

# database_connection.py
import psycopg2
from psycopg2 import sql

class PostgreSQLConnection:
    def __init__(self, host, port, user, password, dbname="postgres"):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.dbname = dbname
        self.conn = None
        self.cursor = None

    def connect(self):
        """Connect to the PostgreSQL instance."""
        try:
            self.conn = psycopg2.connect(
                host=self.host,
                port=self.port,
                user=self.user,
                password=self.password,
                dbname=self.dbname
            )
            self.conn.autocommit = True  # Enable autocommit for database operations
            self.cursor = self.conn.cursor()
            print("Connected to PostgreSQL successfully.")
        except psycopg2.Error as e:
            print(f"An error occurred while connecting: {e}")

    def execute_query(self, query, params=None):
        """Execute a SQL query."""
        try:
            self.cursor.execute(query, params)
            print("Query executed successfully.")
        except psycopg2.Error as e:
            print(f"An error occurred while executing the query: {e}")

    def fetch_data(self, query, params=None):
        """Fetch data from a SQL query."""
        try:
            self.cursor.execute(query, params)
            return self.cursor.fetchall()
        except psycopg2.Error as e:
            print(f"An error occurred while fetching data: {e}")
            return None

    def close(self):
        """Close the connection."""
        if self.conn:
            self.cursor.close()
            self.conn.close()
            print("Connection closed.")

And now, we will use the above to create a Users table with create_user_table.py:

# create_user_table.py
from database_connection import PostgreSQLConnection

class UserTableCreator:
    def __init__(self, connection):
        self.connection = connection

    def create_table(self):
        """Create the 'users' table."""
        create_table_query = """
        CREATE TABLE IF NOT EXISTS users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100) NOT NULL,
            email VARCHAR(100) UNIQUE NOT NULL,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
        """
        self.connection.execute_query(create_table_query)
        print("Table 'users' created or already exists.")

# Example usage
if __name__ == "__main__":
    # Database connection details
    user = "your_user"
    password = "your_password"
    host = "your_host"
    port = "your_port"

    dbname = "my_sample_db"  # Database where the table will be created

    # Connect to the database
    db_connection = PostgreSQLConnection(host, port, user, password, dbname)
    db_connection.connect()

    # Create the 'users' table
    table_creator = UserTableCreator(db_connection)
    table_creator.create_table()

    # Close the connection
    db_connection.close()

Q3: How do I insert synthetic data into the Users table?

Let's create some fake users, of which 8 will be of gmail emails, and rest of them non-gmail.

First install the faker libary:

uv add faker

Then, create create_fake_users.py:

# create_fake_users.py
from database_connection import PostgreSQLConnection
from faker import Faker

class FakeUserInserter:
    def __init__(self, connection):
        self.connection = connection
        self.fake = Faker()

    def generate_fake_users(self, gmail_count=8, non_gmail_count=2):
        """Generate fake users with a mix of Gmail and non-Gmail emails."""
        users = []
        for _ in range(gmail_count):
            name = self.fake.name()
            email = self.fake.email(domain="gmail.com")
            users.append((name, email))

        for _ in range(non_gmail_count):
            name = self.fake.name()
            email = self.fake.email(domain=self.fake.domain_name())
            users.append((name, email))

        return users

    def insert_users(self, users):
        """Insert the generated fake users into the 'users' table."""
        insert_query = """
        INSERT INTO users (name, email) VALUES (%s, %s);
        """
        for user in users:
            self.connection.execute_query(insert_query, user)
        print(f"Inserted {len(users)} users into the 'users' table.")

# Example usage
if __name__ == "__main__":
    # Database connection details
    host = "your_host"
    port = "your_port"
    user = "your_user"
    password = "your_password"
    dbname = "sample_db"  # Database where the table exists

    # Connect to the database
    db_connection = PostgreSQLConnection(host, port, user, password, dbname)
    db_connection.connect()

    # Generate and insert fake users
    fake_user_inserter = FakeUserInserter(db_connection)
    fake_users = fake_user_inserter.generate_fake_users()
    fake_user_inserter.insert_users(fake_users)

    # Close the connection
    db_connection.close()

Q4: How to list the Users Table For Inspection?

Now we will try to list the contents of the Users table in a nice tabular format.

First, let's install tabular library:

uv add tabular

Then, create a file list_users.py:

# list_users.py
from database_connection import PostgreSQLConnection
from tabulate import tabulate

class UserLister:
    def __init__(self, connection):
        self.connection = connection

    def fetch_users(self):
        """Fetch all users from the 'users' table."""
        fetch_query = "SELECT * FROM users;"
        users = self.connection.fetch_data(fetch_query)
        return users

    def display_users(self, users):
        """Display the users in a nice table format."""
        if users:
            headers = ["ID", "Name", "Email", "Updated At"]
            print(tabulate(users, headers=headers, tablefmt="pretty"))
        else:
            print("No users found in the database.")

# Example usage
if __name__ == "__main__":
    # Database connection details
    host = "your_host"
    port = "your_port"
    user = "your_user"
    password = "your_password"
    dbname = "sample_db"  # Database where the table exists

    # Connect to the database
    db_connection = PostgreSQLConnection(host, port, user, password, dbname)
    db_connection.connect()

    # Fetch and display users
    user_lister = UserLister(db_connection)
    users = user_lister.fetch_users()
    user_lister.display_users(users)

    # Close the connection
    db_connection.close()

I get the following result:

Connected to PostgreSQL successfully.
+----+---------------------+------------------------------------+----------------------------+
| ID |        Name         |               Email                |         Updated At         |
+----+---------------------+------------------------------------+----------------------------+
| 1  | Stephanie Henderson |         [e-mail]          | 2025-01-29 18:15:39.922214 |
| 2  |   Joseph Ramirez    |      [e-mail]       | 2025-01-29 18:15:39.953957 |
| 3  |  Molly Bennett DDS  |         [e-mail]         | 2025-01-29 18:15:39.982407 |
| 4  |    Philip Logan     |         [e-mail]         | 2025-01-29 18:15:40.010792 |
| 5  |   Victoria Brown    |       [e-mail]       | 2025-01-29 18:15:40.042211 |
| 6  |    Angela White     |          [e-mail]           | 2025-01-29 18:15:40.071151 |
| 7  |    Ashley Dalton    |      [e-mail]       | 2025-01-29 18:15:40.099926 |
| 8  |  Kenneth Townsend   |         [e-mail]         | 2025-01-29 18:15:40.131158 |
| 9  |   Christine Quinn   | [e-mail] | 2025-01-29 18:15:40.159691 |
| 10 |    Kendra Bryant    |    [e-mail]    | 2025-01-29 18:15:40.190682 |
+----+---------------------+------------------------------------+----------------------------+
Connection closed.