PostgreSQL - Python

PostgreSQL - Modern Python Interface with Psycopg

psycopg is the most popular PostgreSQL database adapter for the Python programming language. This guide focuses on modern practices using Python 3 and the widely-used psycopg2 library. We will also mention its successor, psycopg3, which offers further improvements.

Installation

The recommended way to install psycopg2 is using pip, Python’s package installer. The -binary package includes its own dependencies and is the easiest to get started with.

# It's highly recommended to work inside a virtual environment
python3 -m venv venv
source venv/bin/activate

# Install psycopg2
pip install psycopg2-binary

Note: A newer version, psycopg3, is available and recommended for new projects. It can be installed with pip install psycopg. While the API is similar, this tutorial will use psycopg2 for its vast existing usage.

Connecting to the Database

Best practice is to use a with statement to manage connections and cursors. This ensures that resources are automatically closed even if errors occur. Avoid hardcoding credentials in your code; use environment variables or a configuration management system.

#!/usr/bin/env python3

import psycopg2
import os

# Best Practice: Load credentials from environment variables
DB_NAME = os.getenv("PG_DBNAME", "testdb")
DB_USER = os.getenv("PG_USER", "postgres")
DB_PASS = os.getenv("PG_PASSWORD", "pass123")
DB_HOST = os.getenv("PG_HOST", "127.0.0.1")
DB_PORT = os.getenv("PG_PORT", "5432")

conn_string = f"dbname={DB_NAME} user={DB_USER} password={DB_PASS} host={DB_HOST} port={DB_PORT}"

try:
    # Use 'with' for automatic connection closing
    with psycopg2.connect(conn_string) as conn:
        print("Opened database successfully")
        # The 'with' block also handles conn.commit() on success
        # or conn.rollback() on error.
except psycopg2.OperationalError as e:
    print(f"Could not connect to the database: {e}")

Executing Queries: CRUD Operations

Crucial Security Note: Always use parameterized queries to pass values into your SQL. Never use f-strings or string concatenation to build queries with user-provided data, as this exposes you to SQL injection attacks. psycopg2 uses %s as a placeholder, regardless of the data type.

CREATE TABLE

#!/usr/bin/env python3
import psycopg2

# (Use connection string from the previous example)

create_table_query = """
CREATE TABLE IF NOT EXISTS employees (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    age INTEGER NOT NULL,
    department TEXT,
    salary NUMERIC(10, 2)
);
"""

try:
    with psycopg2.connect(conn_string) as conn:
        with conn.cursor() as cur:
            cur.execute(create_table_query)
            print("Table 'employees' created or already exists.")
except (psycopg2.Error) as error:
    print(f"Error while connecting to PostgreSQL: {error}")

INSERT Operation (Parameterized)

insert_query = "INSERT INTO employees (name, age, department, salary) VALUES (%s, %s, %s, %s) RETURNING id;"
new_employee = ('Alice', 30, 'Engineering', 95000.00)

try:
    with psycopg2.connect(conn_string) as conn:
        with conn.cursor() as cur:
            cur.execute(insert_query, new_employee)
            inserted_id = cur.fetchone()[0]
            print(f"Record inserted successfully with ID: {inserted_id}")
except (psycopg2.Error) as error:
    print(f"Error during insert: {error}")

SELECT Operation

For more readable code, you can use psycopg2.extras.DictCursor to get results as dictionary-like objects instead of tuples.

from psycopg2.extras import DictCursor

select_query = "SELECT id, name, department, salary FROM employees WHERE department = %s;"

try:
    with psycopg2.connect(conn_string) as conn:
        # Use a DictCursor to access columns by name
        with conn.cursor(cursor_factory=DictCursor) as cur:
            cur.execute(select_query, ('Engineering',))
            rows = cur.fetchall()
            print(f"Found {cur.rowcount} employees in Engineering:")
            for row in rows:
                print(f"  ID: {row['id']}, Name: {row['name']}, Salary: {row['salary']}")
except (psycopg2.Error) as error:
    print(f"Error during select: {error}")

UPDATE and DELETE Operations

update_query = "UPDATE employees SET salary = %s WHERE name = %s;"
delete_query = "DELETE FROM employees WHERE name = %s;"

try:
    with psycopg2.connect(conn_string) as conn:
        with conn.cursor() as cur:
            # UPDATE
            cur.execute(update_query, (100000.00, 'Alice'))
            print(f"Rows updated: {cur.rowcount}")

            # DELETE
            cur.execute(delete_query, ('Bob',))
            print(f"Rows deleted: {cur.rowcount}")
except (psycopg2.Error) as error:
    print(f"Error during update/delete: {error}")