Python PostgreSQL

PostgreSQL is a powerful, open-source object-relational database system with over 30 years of active development. In this tutorial, you will learn how to connect Python applications to PostgreSQL databases and perform various operations using the psycopg2 library.

Introduction

PostgreSQL (often called "Postgres") offers many advanced features like complex queries, foreign keys, triggers, views, transactional integrity, and multiversion concurrency control. Learning to connect your Python applications to PostgreSQL opens up these powerful database capabilities to your programs.

This guide covers:

Setting up the PostgreSQL connection
CRUD operations (Create, Read, Update, Delete)
Working with transactions
Advanced features and best practices

Prerequisites

Before getting started, make sure you have:

Python installed on your machine
PostgreSQL server installed and running
Basic knowledge of SQL
The psycopg2 library installed

You can install the psycopg2 library using pip:

bash
pip install psycopg2

If you encounter issues with the binary package, you can try the binary version:

bash
pip install psycopg2-binary

Connecting to PostgreSQL

The first step to working with PostgreSQL in Python is establishing a connection to the database.

python
import psycopg2

# Connection parameters
conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

# Establishing a connection
try:
    conn = psycopg2.connect(**conn_params)
    print("Connection established successfully!")
    
    # Create a cursor
    cursor = conn.cursor()
    
    # Execute some test SQL
    cursor.execute("SELECT version();")
    
    # Fetch the result
    db_version = cursor.fetchone()
    print(f"PostgreSQL database version: {db_version[0]}")
    
    # Close the cursor and connection
    cursor.close()
    conn.close()
    print("PostgreSQL connection is closed")
    
except (Exception, psycopg2.Error) as error:
    print(f"Error while connecting to PostgreSQL: {error}")

Output:

Connection established successfully!
PostgreSQL database version: PostgreSQL 14.5 on x86_64-pc-linux-gnu, compiled by gcc (GCC) 11.2.0, 64-bit
PostgreSQL connection is closed

Understanding Connection Parameters

host: The server where PostgreSQL is running (default: localhost)
database: The name of the database you want to connect to
user: Username for authentication
password: Password for authentication
port: The port PostgreSQL is running on (default: 5432)

Using Connection Context Manager

A better way to handle connections is using Python's context manager (with statement):

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT version();")
            db_version = cursor.fetchone()
            print(f"PostgreSQL database version: {db_version[0]}")
            
except (Exception, psycopg2.Error) as error:
    print(f"Error while connecting to PostgreSQL: {error}")

Creating Tables

Let's create a users table to use for our examples:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

create_table_query = '''
CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    created_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    last_login TIMESTAMP
);
'''

try:
    with psycopg2.connect(**conn_params) as conn:
        # Autocommit mode
        conn.autocommit = True
        
        with conn.cursor() as cursor:
            cursor.execute(create_table_query)
            print("Table created successfully in PostgreSQL")
            
except (Exception, psycopg2.Error) as error:
    print(f"Error while creating table: {error}")

CRUD Operations with PostgreSQL

Let's explore the four basic database operations: Create, Read, Update, and Delete.

Create (Insert) Operation

Insert a single record into the users table:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

insert_query = '''
INSERT INTO users (username, email)
VALUES (%s, %s) RETURNING id;
'''

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute(insert_query, ('john_doe', '[email protected]'))
            
            # Get the generated id
            user_id = cursor.fetchone()[0]
            print(f"User inserted with ID: {user_id}")
            
            # Commit the transaction
            conn.commit()
            
except (Exception, psycopg2.Error) as error:
    print(f"Error while inserting data: {error}")

Output:

User inserted with ID: 1

Inserting Multiple Records

You can also insert multiple records at once:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

insert_query = '''
INSERT INTO users (username, email)
VALUES (%s, %s);
'''

users_to_insert = [
    ('jane_doe', '[email protected]'),
    ('bob_smith', '[email protected]'),
    ('alice_johnson', '[email protected]')
]

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.executemany(insert_query, users_to_insert)
            print(f"Successfully inserted {cursor.rowcount} records")
            
            # Commit the transaction
            conn.commit()
            
except (Exception, psycopg2.Error) as error:
    print(f"Error while inserting data: {error}")

Output:

Successfully inserted 3 records

Read (Select) Operation

Fetch records from the database:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb", 
    "user": "postgres",
    "password": "password"
}

select_query = "SELECT id, username, email FROM users;"

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute(select_query)
            
            print("All users:")
            records = cursor.fetchall()
            
            for row in records:
                print(f"ID: {row[0]}, Username: {row[1]}, Email: {row[2]}")
                
except (Exception, psycopg2.Error) as error:
    print(f"Error while fetching data: {error}")

Output:

All users:
ID: 1, Username: john_doe, Email: [email protected]
ID: 2, Username: jane_doe, Email: [email protected]
ID: 3, Username: bob_smith, Email: [email protected]
ID: 4, Username: alice_johnson, Email: [email protected]

Update Operation

Let's update user information:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

update_query = '''
UPDATE users
SET email = %s
WHERE username = %s;
'''

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute(update_query, ('[email protected]', 'john_doe'))
            
            # Commit the transaction
            conn.commit()
            
            print(f"Successfully updated {cursor.rowcount} record(s)")
            
except (Exception, psycopg2.Error) as error:
    print(f"Error while updating data: {error}")

Output:

Successfully updated 1 record(s)

Delete Operation

Remove records from the database:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

delete_query = "DELETE FROM users WHERE username = %s;"

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute(delete_query, ('bob_smith',))
            
            # Commit the transaction
            conn.commit()
            
            print(f"Successfully deleted {cursor.rowcount} record(s)")
            
except (Exception, psycopg2.Error) as error:
    print(f"Error while deleting data: {error}")

Output:

Successfully deleted 1 record(s)

Working with Transactions

Transactions allow you to execute a series of operations as an atomic unit. If any operation fails, all operations can be rolled back.

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

try:
    with psycopg2.connect(**conn_params) as conn:
        # Turn autocommit off (it's off by default)
        conn.autocommit = False
        
        with conn.cursor() as cursor:
            # First operation
            cursor.execute("UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE username = %s;", ('john_doe',))
            
            # Second operation
            cursor.execute("INSERT INTO users (username, email) VALUES (%s, %s);", ('new_user', '[email protected]'))
            
            # If everything is successful, commit the transaction
            conn.commit()
            print("Transaction completed successfully!")
            
except (Exception, psycopg2.Error) as error:
    # If any operation fails, roll back the transaction
    if conn:
        conn.rollback()
    print(f"Transaction failed. Error: {error}")

Advanced Features

Using Named Cursors for Large Results

When dealing with large result sets, you can use named cursors to fetch records in batches:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

try:
    with psycopg2.connect(**conn_params) as conn:
        # Create a named cursor
        with conn.cursor(name='large_result') as cursor:
            cursor.execute("SELECT * FROM users")
            
            # Fetch in batches of 2 records
            batch_size = 2
            records = cursor.fetchmany(batch_size)
            
            batch_num = 1
            while records:
                print(f"Batch {batch_num}:")
                for row in records:
                    print(f"  {row}")
                    
                records = cursor.fetchmany(batch_size)
                batch_num += 1
                
except (Exception, psycopg2.Error) as error:
    print(f"Error: {error}")

Handling NULL Values

When working with databases, NULL values require special handling:

python
import psycopg2

conn_params = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password"
}

try:
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT username, last_login FROM users")
            
            for username, last_login in cursor.fetchall():
                if last_login is None:
                    print(f"{username} has never logged in")
                else:
                    print(f"{username} last logged in at {last_login}")
                    
except (Exception, psycopg2.Error) as error:
    print(f"Error: {error}")

Real-World Example: User Authentication System

Let's build a simple user authentication system using PostgreSQL:

python
import psycopg2
import hashlib
import os

class UserAuth:
    def __init__(self, conn_params):
        self.conn_params = conn_params
        self._create_tables()
        
    def _create_tables(self):
        create_table_query = '''
        CREATE TABLE IF NOT EXISTS auth_users (
            id SERIAL PRIMARY KEY,
            username VARCHAR(50) UNIQUE NOT NULL,
            password_hash VARCHAR(128) NOT NULL,
            salt VARCHAR(32) NOT NULL,
            email VARCHAR(100) UNIQUE NOT NULL,
            is_active BOOLEAN DEFAULT TRUE,
            created_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            last_login TIMESTAMP
        );
        '''
        
        try:
            with psycopg2.connect(**self.conn_params) as conn:
                conn.autocommit = True
                with conn.cursor() as cursor:
                    cursor.execute(create_table_query)
                    
        except (Exception, psycopg2.Error) as error:
            print(f"Error while creating table: {error}")
            
    def _hash_password(self, password, salt=None):
        if salt is None:
            salt = os.urandom(16).hex()  # 16 bytes = 32 hex characters
            
        # Create hash with the salt
        hash_obj = hashlib.sha256(salt.encode() + password.encode())
        password_hash = hash_obj.hexdigest()
        
        return password_hash, salt
        
    def register_user(self, username, password, email):
        password_hash, salt = self._hash_password(password)
        
        insert_query = '''
        INSERT INTO auth_users (username, password_hash, salt, email)
        VALUES (%s, %s, %s, %s) RETURNING id;
        '''
        
        try:
            with psycopg2.connect(**self.conn_params) as conn:
                with conn.cursor() as cursor:
                    cursor.execute(insert_query, (username, password_hash, salt, email))
                    user_id = cursor.fetchone()[0]
                    conn.commit()
                    return user_id
                    
        except (Exception, psycopg2.Error) as error:
            print(f"Error while registering user: {error}")
            return None
            
    def authenticate_user(self, username, password):
        select_query = '''
        SELECT id, password_hash, salt, is_active FROM auth_users
        WHERE username = %s;
        '''
        
        update_query = '''
        UPDATE auth_users SET last_login = CURRENT_TIMESTAMP
        WHERE id = %s;
        '''
        
        try:
            with psycopg2.connect(**self.conn_params) as conn:
                with conn.cursor() as cursor:
                    cursor.execute(select_query, (username,))
                    user = cursor.fetchone()
                    
                    if not user:
                        return False, "User not found"
                        
                    user_id, stored_hash, salt, is_active = user
                    
                    if not is_active:
                        return False, "Account is disabled"
                    
                    # Hash the provided password with the stored salt
                    calculated_hash, _ = self._hash_password(password, salt)
                    
                    if calculated_hash == stored_hash:
                        # Update last login time
                        cursor.execute(update_query, (user_id,))
                        conn.commit()
                        return True, user_id
                    else:
                        return False, "Incorrect password"
                        
        except (Exception, psycopg2.Error) as error:
            print(f"Error during authentication: {error}")
            return False, "Database error"

# Usage example
if __name__ == "__main__":
    conn_params = {
        "host": "localhost",
        "database": "mydb",
        "user": "postgres",
        "password": "password"
    }
    
    auth = UserAuth(conn_params)
    
    # Register a new user
    user_id = auth.register_user("test_user", "secure_password", "[email protected]")
    print(f"Registered user with ID: {user_id}")
    
    # Authenticate with correct credentials
    success, result = auth.authenticate_user("test_user", "secure_password")
    if success:
        print(f"Authentication successful! User ID: {result}")
    else:
        print(f"Authentication failed: {result}")
    
    # Authenticate with incorrect credentials
    success, result = auth.authenticate_user("test_user", "wrong_password")
    if success:
        print(f"Authentication successful! User ID: {result}")
    else:
        print(f"Authentication failed: {result}")

Output:

Registered user with ID: 1
Authentication successful! User ID: 1
Authentication failed: Incorrect password

Best Practices

When working with PostgreSQL in Python, keep these best practices in mind:

Always use connection pooling for production applications (using libraries like psycopg2.pool or SQLAlchemy)
Use parameterized queries to prevent SQL injection attacks
Handle database connections properly using context managers
Set appropriate timeouts for database operations
Implement proper error handling to gracefully handle database errors
Use transactions for operations that need to be atomic
Close connections when they're no longer needed
Use environment variables for database credentials, not hardcoded values
Create indexes for frequently queried columns
Monitor database performance in production environments

Summary

In this tutorial, you learned how to:

Connect to a PostgreSQL database using Python's psycopg2 library
Perform CRUD operations (Create, Read, Update, Delete)
Work with transactions
Implement a real-world example (user authentication system)
Follow best practices for working with PostgreSQL

PostgreSQL with Python gives you a powerful combination for building data-driven applications, from simple CRUD operations to complex enterprise systems.

Additional Resources

Psycopg2 Documentation
PostgreSQL Official Documentation
PostgreSQL Tutorial
SQLAlchemy ORM - A higher-level Python SQL toolkit and ORM

Exercises

Create a simple blog application with tables for posts, comments, and users
Implement a product inventory system with tables for products, categories, and inventory levels
Build a todo list application that stores tasks in a PostgreSQL database
Create a function that backs up a table to a CSV file
Implement a simple database migration system for schema changes

These exercises will help you gain practical experience with PostgreSQL in Python applications.

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction​

Prerequisites​

Connecting to PostgreSQL​

Understanding Connection Parameters​

Using Connection Context Manager​

Creating Tables​

CRUD Operations with PostgreSQL​

Create (Insert) Operation​

Inserting Multiple Records​

Read (Select) Operation​

Update Operation​

Delete Operation​

Working with Transactions​

Advanced Features​

Using Named Cursors for Large Results​

Handling NULL Values​

Real-World Example: User Authentication System​

Best Practices​

Summary​

Additional Resources​

Exercises​

Introduction

Prerequisites

Connecting to PostgreSQL

Understanding Connection Parameters

Using Connection Context Manager

Creating Tables

CRUD Operations with PostgreSQL

Create (Insert) Operation

Inserting Multiple Records

Read (Select) Operation

Update Operation

Delete Operation

Working with Transactions

Advanced Features

Using Named Cursors for Large Results

Handling NULL Values

Real-World Example: User Authentication System

Best Practices

Summary

Additional Resources

Exercises