Connect Python applications to PostgreSQL databases using the powerful Psycopg2 driver
Connect Python applications to PostgreSQL databases using the powerful Psycopg2 driver
Python's flexibility and PostgreSQL's robust feature set make a powerful combination for data-driven applications. In this lesson, you'll learn how to connect Python applications to PostgreSQL databases using Psycopg2, the most popular PostgreSQL adapter for Python.
Psycopg2 is a PostgreSQL database adapter for Python. It's designed to be:
By the end of this lesson, you'll be able to integrate PostgreSQL with your Python applications confidently, using best practices for security, performance, and maintainability.
Let's get started with setting up the environment.
Before we start writing code, let's set up our environment with Python and the Psycopg2 library.
Psycopg2 has two versions available:
For development and learning purposes, the binary package is easiest:
1# Install the binary package
2pip install psycopg2-binary
3
4# Or for production environments, install from source
5pip install psycopg2If you encounter installation issues with the standard package, you might need to install PostgreSQL development libraries first:
1# Ubuntu/Debian
2sudo apt-get install python3-dev libpq-dev
3
4# Red Hat/Fedora
5sudo dnf install python3-devel postgresql-devel
6
7# macOS (using Homebrew)
8brew install postgresqlLet's verify that Psycopg2 is installed correctly by creating a simple test script:
1import psycopg2
2
3# Print the version
4print(psycopg2.__version__)
5
6# Try connecting to PostgreSQL
7try:
8 conn = psycopg2.connect(
9 host="localhost",
10 database="postgres",
11 user="postgres",
12 password="your_password" # Replace with your actual password
13 )
14 print("Connection successful!")
15 conn.close()
16except Exception as e:
17 print(f"Connection failed: {e}")If you see the version number and "Connection successful!" message, you're ready to proceed. If not, check your PostgreSQL installation and connection details.
For this lesson, we'll create a simple "bookstore" database to work with:
1-- Connect to PostgreSQL and run these commands
2CREATE DATABASE bookstore;
3
4\c bookstore
5
6CREATE TABLE books (
7 id SERIAL PRIMARY KEY,
8 title VARCHAR(100) NOT NULL,
9 author VARCHAR(100) NOT NULL,
10 price DECIMAL(10,2) NOT NULL,
11 stock INTEGER NOT NULL DEFAULT 0
12);
13
14INSERT INTO books (title, author, price, stock) VALUES
15 ('The Great Gatsby', 'F. Scott Fitzgerald', 12.99, 25),
16 ('To Kill a Mockingbird', 'Harper Lee', 14.99, 30),
17 ('1984', 'George Orwell', 11.99, 20),
18 ('Pride and Prejudice', 'Jane Austen', 9.99, 15),
19 ('The Catcher in the Rye', 'J.D. Salinger', 13.99, 22);With our environment set up and test database ready, we can now start exploring how to connect Python to PostgreSQL using Psycopg2.
Now that we have our environment set up, let's explore the fundamental database operations using Psycopg2.
The first step is to establish a connection to the PostgreSQL database:
1import psycopg2
2
3def get_connection():
4 conn = psycopg2.connect(
5 host="localhost",
6 database="bookstore",
7 user="postgres",
8 password="your_password" # Use environment variables in production!
9 )
10 return conn> Security Best Practice: Never hardcode passwords in your code. Use environment variables or a configuration file that's not checked into version control.
1import os
2import psycopg2
3
4def get_connection_secure():
5 conn = psycopg2.connect(
6 host=os.environ.get("DB_HOST", "localhost"),
7 database=os.environ.get("DB_NAME", "bookstore"),
8 user=os.environ.get("DB_USER", "postgres"),
9 password=os.environ.get("DB_PASSWORD")
10 )
11 return connOnce connected, you can create a cursor object to execute SQL commands:
1def fetch_all_books():
2 conn = get_connection()
3 cur = conn.cursor()
4
5 # Execute a query
6 cur.execute("SELECT id, title, author, price, stock FROM books")
7
8 # Fetch all results
9 books = cur.fetchall()
10
11 # Close cursor and connection
12 cur.close()
13 conn.close()
14
15 return books
16
17# Example usage
18books = fetch_all_books()
19for book in books:
20 print(f"ID: {book[0]}, Title: {book[1]}, Author: {book[2]}, Price: ${book[3]}, Stock: {book[4]}")Python's context managers make it easier to manage database connections and ensure proper cleanup:
1def fetch_books_with_context():
2 with get_connection() as conn:
3 with conn.cursor() as cur:
4 cur.execute("SELECT id, title, author, price, stock FROM books")
5 books = cur.fetchall()
6
7 return books # Connection is automatically closedNEVER use string concatenation to build SQL queries. Use parameterized queries instead to prevent SQL injection:
1def find_books_by_author(author):
2 with get_connection() as conn:
3 with conn.cursor() as cur:
4 # Unsafe (DO NOT USE):
5 # cur.execute(f"SELECT * FROM books WHERE author = '{author}'")
6
7 # Safe (USE THIS):
8 cur.execute("SELECT * FROM books WHERE author = %s", (author,))
9 return cur.fetchall()
10
11# Find books by Harper Lee
12books = find_books_by_author("Harper Lee")1def add_new_book(title, author, price, stock):
2 with get_connection() as conn:
3 with conn.cursor() as cur:
4 cur.execute(
5 "INSERT INTO books (title, author, price, stock) VALUES (%s, %s, %s, %s) RETURNING id",
6 (title, author, price, stock)
7 )
8 # Get the ID of the newly inserted book
9 book_id = cur.fetchone()[0]
10
11 # Commit the transaction
12 conn.commit()
13
14 return book_id
15
16# Add a new book
17new_book_id = add_new_book("The Hobbit", "J.R.R. Tolkien", 15.99, 35)
18print(f"Added new book with ID: {new_book_id}")1def update_book_price(book_id, new_price):
2 with get_connection() as conn:
3 with conn.cursor() as cur:
4 cur.execute(
5 "UPDATE books SET price = %s WHERE id = %s",
6 (new_price, book_id)
7 )
8 rows_updated = cur.rowcount
9 conn.commit()
10 return rows_updated
11
12# Update a book's price
13updated = update_book_price(1, 13.99)
14print(f"Updated {updated} book(s)")1def delete_book(book_id):
2 with get_connection() as conn:
3 with conn.cursor() as cur:
4 cur.execute("DELETE FROM books WHERE id = %s", (book_id,))
5 rows_deleted = cur.rowcount
6 conn.commit()
7 return rows_deleted
8
9# Delete a book
10deleted = delete_book(5)
11print(f"Deleted {deleted} book(s)")By default, Psycopg2 returns results as tuples. You can change this behavior:
1import psycopg2.extras
2
3def fetch_books_as_dict():
4 with get_connection() as conn:
5 # Use DictCursor to get results as dictionaries
6 with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
7 cur.execute("SELECT * FROM books")
8 books = cur.fetchall()
9
10 # Books is a list of dictionary-like objects
11 return books
12
13books = fetch_books_as_dict()
14for book in books:
15 print(f"Title: {book['title']}, Author: {book['author']}")These basic operations form the foundation of working with PostgreSQL in Python. In the next section, we'll explore more advanced techniques for managing transactions and handling errors.
Proper transaction management and error handling are essential for building robust database applications. Let's explore how to implement these with Psycopg2.
In PostgreSQL, all database operations occur within a transaction. By default, Psycopg2 doesn't auto-commit changes - you need to explicitly call conn.commit() to save changes to the database.
1def transfer_book_stock(from_id, to_id, amount):
2 conn = get_connection()
3 try:
4 cur = conn.cursor()
5
6 # Decrease stock from the source book
7 cur.execute("UPDATE books SET stock = stock - %s WHERE id = %s", (amount, from_id))
8
9 # Increase stock for the destination book
10 cur.execute("UPDATE books SET stock = stock + %s WHERE id = %s", (amount, to_id))
11
12 # Commit the transaction - this makes the changes permanent
13 conn.commit()
14 print("Stock transfer completed successfully!")
15
16 except Exception as e:
17 # If any error occurs, roll back the transaction
18 conn.rollback()
19 print(f"Error: {e}")
20 raise
21
22 finally:
23 # Always close the connection
24 conn.close()Context managers make transaction handling cleaner:
1def transfer_stock_with_context(from_id, to_id, amount):
2 try:
3 with get_connection() as conn:
4 with conn.cursor() as cur:
5 # Decrease stock from the source book
6 cur.execute("UPDATE books SET stock = stock - %s WHERE id = %s",
7 (amount, from_id))
8
9 # Increase stock for the destination book
10 cur.execute("UPDATE books SET stock = stock + %s WHERE id = %s",
11 (amount, to_id))
12
13 # The transaction is automatically committed when the context exits
14
15 print("Stock transfer completed successfully!")
16
17 except Exception as e:
18 # The transaction is automatically rolled back on error
19 print(f"Error: {e}")
20 raisePsycopg2 exposes PostgreSQL error codes, which allows you to handle specific database errors:
1from psycopg2 import errors
2
3def withdraw_book_stock(book_id, amount):
4 try:
5 with get_connection() as conn:
6 with conn.cursor() as cur:
7 # Check current stock
8 cur.execute("SELECT stock FROM books WHERE id = %s", (book_id,))
9 current_stock = cur.fetchone()[0]
10
11 if current_stock < amount:
12 raise ValueError("Not enough stock available")
13
14 # Update stock
15 cur.execute("UPDATE books SET stock = stock - %s WHERE id = %s",
16 (amount, book_id))
17 conn.commit()
18
19 return True
20
21 except errors.CheckViolation as e:
22 # This catches PostgreSQL check constraint violations
23 print(f"Check constraint violated: {e}")
24 return False
25
26 except errors.ForeignKeyViolation as e:
27 # This catches foreign key violations
28 print(f"Foreign key constraint violated: {e}")
29 return False
30
31 except errors.UniqueViolation as e:
32 # This catches unique constraint violations
33 print(f"Unique constraint violated: {e}")
34 return False
35
36 except ValueError as e:
37 # Application-level validation
38 print(f"Validation error: {e}")
39 return False
40
41 except Exception as e:
42 # Catch-all for other errors
43 print(f"Unexpected error: {e}")
44 return FalseFor complex operations, you might need savepoints to create "checkpoints" within a transaction:
1def complex_inventory_operation(book_id):
2 try:
3 with get_connection() as conn:
4 with conn.cursor() as cur:
5 # Start a transaction
6
7 # First operation: Update price
8 cur.execute("UPDATE books SET price = price * 1.1 WHERE id = %s", (book_id,))
9
10 # Create a savepoint
11 cur.execute("SAVEPOINT price_updated")
12
13 try:
14 # Second operation: Update stock
15 cur.execute("UPDATE books SET stock = stock - 5 WHERE id = %s", (book_id,))
16
17 # Check if stock went negative
18 cur.execute("SELECT stock FROM books WHERE id = %s", (book_id,))
19 new_stock = cur.fetchone()[0]
20
21 if new_stock < 0:
22 # Roll back to the savepoint
23 cur.execute("ROLLBACK TO SAVEPOINT price_updated")
24 print("Stock update rolled back, but price update kept")
25 else:
26 # Everything is good, continue
27 print("Both price and stock updated")
28
29 except Exception as e:
30 # Roll back to the savepoint
31 cur.execute("ROLLBACK TO SAVEPOINT price_updated")
32 print(f"Stock update failed, rolled back to savepoint: {e}")
33
34 # Commit what's left of our transaction
35 conn.commit()
36
37 except Exception as e:
38 print(f"Entire transaction failed: {e}")
39 # No need to rollback here as the context manager does itPsycopg2 allows you to set the transaction isolation level for fine-grained control:
1def read_with_isolation_level(isolation_level):
2 conn = get_connection()
3 try:
4 # Set isolation level
5 conn.set_isolation_level(isolation_level)
6
7 with conn.cursor() as cur:
8 cur.execute("SELECT * FROM books")
9 return cur.fetchall()
10
11 finally:
12 conn.close()
13
14# Import isolation levels
15from psycopg2.extensions import ISOLATION_LEVEL_READ_COMMITTED
16from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE
17
18# Read with different isolation levels
19books_rc = read_with_isolation_level(ISOLATION_LEVEL_READ_COMMITTED)
20books_s = read_with_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)When dealing with large result sets, server-side cursors prevent loading everything into memory at once:
1def process_large_result_set():
2 conn = get_connection()
3 try:
4 # Name the cursor to make it server-side
5 with conn.cursor(name='large_result_cursor') as cur:
6 cur.execute("SELECT * FROM books")
7
8 # Fetch and process in batches of 100
9 batch_size = 100
10 while True:
11 records = cur.fetchmany(batch_size)
12 if not records:
13 break
14
15 for record in records:
16 # Process each record
17 print(f"Processing: {record}")
18
19 finally:
20 conn.close()Proper transaction and error handling are critical for data integrity. In the next section, we'll look at connection pooling and advanced patterns for production applications.
:
1from psycopg2 import pool
2import threading
3
4# Create a threaded connection pool
5threaded_pool = pool.ThreadedConnectionPool(
6 minconn=1,
7 maxconn=20,
8 host="localhost",
9 database="bookstore",
10 user="postgres",
11 password="your_password"
12)
13
14# Example with threading
15def process_book(book_id):
16 conn = threaded_pool.getconn(key=threading.current_thread())
17 try:
18 with conn.cursor() as cur:
19 cur.execute("SELECT * FROM books WHERE id = %s", (book_id,))
20 book = cur.fetchone()
21 print(f"Thread {threading.current_thread().name} processed book: {book}")
22 finally:
23 threaded_pool.putconn(conn, key=threading.current_thread())
24
25# Create and start threads
26threads = []
27for i in range(1, 6):
28 t = threading.Thread(target=process_book, args=(i,))
29 threads.append(t)
30 t.start()
31
32# Wait for all threads to complete
33for t in threads:
34 t.join()
35
36# Close all connections in the pool when done
37threaded_pool.closeall()You can customize connection behavior with a connection factory:
1import psycopg2
2import psycopg2.extras
3
4def dict_connection_factory(*args, **kwargs):
5 connection = psycopg2.connect(*args, **kwargs)
6
7 # Set this connection to always use DictCursor
8 psycopg2.extensions.register_type(
9 psycopg2.extensions.UNICODE, connection)
10
11 # Set default cursor factory
12 connection.cursor_factory = psycopg2.extras.DictCursor
13
14 return connection
15
16# Create a connection using the factory
17conn = dict_connection_factory(
18 host="localhost",
19 database="bookstore",
20 user="postgres",
21 password="your_password"
22)
23
24# Now all cursors from this connection will return dict-like rows
25with conn.cursor() as cur:
26 cur.execute("SELECT * FROM books LIMIT 3")
27 books = cur.fetchall()
28 for book in books:
29 # We can access columns by name
30 print(f"Title: {book['title']}, Author: {book['author']}")
31
32conn.close()For larger applications, a repository pattern provides a clean abstraction over database operations:
1class BookRepository:
2 def __init__(self, connection_pool):
3 self.connection_pool = connection_pool
4
5 def find_by_id(self, book_id):
6 with self.connection_pool.get_connection() as conn:
7 with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
8 cur.execute("SELECT * FROM books WHERE id = %s", (book_id,))
9 return cur.fetchone()
10
11 def find_all(self):
12 with self.connection_pool.get_connection() as conn:
13 with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
14 cur.execute("SELECT * FROM books")
15 return cur.fetchall()
16
17 def save(self, book_data):
18 with self.connection_pool.get_connection() as conn:
19 with conn.cursor() as cur:
20 if 'id' in book_data and book_data['id']:
21 # Update existing book
22 cur.execute("""
23 UPDATE books
24 SET title = %s, author = %s, price = %s, stock = %s
25 WHERE id = %s
26 RETURNING id
27 """, (
28 book_data['title'],
29 book_data['author'],
30 book_data['price'],
31 book_data['stock'],
32 book_data['id']
33 ))
34 else:
35 # Insert new book
36 cur.execute("""
37 INSERT INTO books (title, author, price, stock)
38 VALUES (%s, %s, %s, %s)
39 RETURNING id
40 """, (
41 book_data['title'],
42 book_data['author'],
43 book_data['price'],
44 book_data['stock']
45 ))
46
47 return cur.fetchone()[0]
48
49 def delete(self, book_id):
50 with self.connection_pool.get_connection() as conn:
51 with conn.cursor() as cur:
52 cur.execute("DELETE FROM books WHERE id = %s", (book_id,))
53 return cur.rowcount > 0
54
55# Example usage
56pool = ConnectionPool(**get_db_config())
57book_repo = BookRepository(pool)
58
59# Find a book
60book = book_repo.find_by_id(1)
61print(book)
62
63# Update a book
64book['price'] = 14.99
65book_id = book_repo.save(book)
66print(f"Updated book ID: {book_id}")
67
68# Add a new book
69new_book = {
70 'title': 'Python Database Programming',
71 'author': 'DB Expert',
72 'price': 29.99,
73 'stock': 50
74}
75new_book_id = book_repo.save(new_book)
76print(f"New book ID: {new_book_id}")These advanced patterns will help you build robust, scalable applications with PostgreSQL and Python. In the next section, we'll build a practical example that ties all these concepts together.
Your progress is saved in this browser, on this device only.