python psycopg connect postgresql and execute queries: Pract

python psycopg connect postgresql and execute queries requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

When you need to connect to a PostgreSQL database from Python and run queries, psycopg is the standard library. This article shows how to use psycopg to connect, execute queries, handle transactions, and manage errors safely. The examples use psycopg3, the modern version of the library, but the core concepts apply to psycopg2 as well.

Installing psycopg and Creating a Connection

Install psycopg with pip:

pip install psycopg

For psycopg2, the package name is psycopg2-binary and the import is psycopg2. The rest of the API is similar, but psycopg3 introduces several improvements, including better type handling and a cleaner connection context manager.

Create a connection using keyword arguments or a connection string:

import psycopg

conn = psycopg.connect(
    host="localhost",
    port=5432,
    dbname="mydb",
    user="postgres",
    password="secret"
)

Alternatively, use a connection string:

conn = psycopg.connect("postgresql://postgres:secret@localhost:5432/mydb")

The connection object represents a single database session. It is not thread-safe by default; use a separate connection per thread or rely on a connection pool. Always close the connection when done, or use a context manager to handle it automatically.

Executing Queries with a Cursor

A cursor is the object that executes SQL and fetches results. Obtain a cursor from the connection and call execute():

cur = conn.cursor()
cur.execute("SELECT id, name FROM users")
rows = cur.fetchall()

fetchall() returns a list of tuples. For a single row, use fetchone(). To iterate without loading everything into memory, use the cursor as an iterator:

cur.execute("SELECT id, name FROM users")
for row in cur:
    print(row)

Cursors are lightweight and should be closed after use. Use the cursor as a context manager to ensure cleanup:

with conn.cursor() as cur:
    cur.execute("SELECT 1")
    print(cur.fetchone())

Using Parameterized Queries to Prevent SQL Injection

Never build SQL by concatenating strings. Use placeholders and pass parameters to execute():

cur.execute(
    "SELECT id, name FROM users WHERE email = %s",
    (email,)
)

The %s placeholder works for all types. Psycopg adapts the Python value to the correct PostgreSQL type. This prevents SQL injection and handles quoting automatically. For named parameters, use %(name)s and pass a dictionary:

cur.execute(
    "SELECT id FROM users WHERE name = %(name)s AND age > %(age)s",
    {"name": "Alice", "age": 30}
)

Parameterized queries also improve performance when the same query is executed repeatedly with different values, as the database can reuse the query plan.

Managing Transactions and Autocommit

By default, psycopg opens a transaction when the first statement is executed. You must commit or roll back to end it. The connection context manager handles this for you:

with psycopg.connect("postgresql://...") as conn:
    with conn.cursor() as cur:
        cur.execute("INSERT INTO users (name) VALUES (%s)", ("Bob",))
    # If the block completes, conn.commit() is called.
    # If an exception occurs, conn.rollback() is called.

For read-only operations or when you want each statement to commit immediately, set autocommit=True:

conn = psycopg.connect("postgresql://...", autocommit=True)

Autocommit is useful for CREATE DATABASE or other commands that cannot run inside a transaction. For most applications, keep the default and use explicit commit/rollback.

Handling Errors and Cleaning Up Resources

Psycopg raises exceptions that inherit from psycopg.Error. Catch them to handle database failures gracefully:

from psycopg import OperationalError, IntegrityError

try:
    with psycopg.connect("postgresql://...") as conn:
        with conn.cursor() as cur:
            cur.execute("INSERT INTO users (id) VALUES (%s)", (1,))
except IntegrityError as e:
    print(f"Duplicate key: {e}")
except OperationalError as e:
    print(f"Connection failed: {e}")

Always close connections and cursors, even on errors. The context managers handle this automatically. If you manage resources manually, use try/finally to ensure cleanup.

Improving Performance with Server-Side Cursors and Connection Pooling

For queries that return a large number of rows, a client-side cursor loads all rows into memory. Use a server-side cursor (named cursor) to stream rows in batches:

with conn.cursor(name="large_query") as cur:
    cur.itersize = 1000
    cur.execute("SELECT * FROM events")
    for row in cur:
        process(row)

The named cursor keeps the result set on the server and fetches rows incrementally. This reduces memory usage on the client.

For applications that create many connections, use a connection pool. The psycopg_pool package provides ConnectionPool:

from psycopg_pool import ConnectionPool

pool = ConnectionPool("postgresql://...", min_size=2, max_size=10)
with pool.connection() as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT 1")

Pooling avoids the overhead of establishing a new connection for every request. Choose a pool size based on your database's connection limits and workload.

Using Row Factories for Dictionary Access

By default, rows are tuples. To access columns by name, use a row factory. Psycopg3 provides dict_row:

from psycopg.rows import dict_row

with psycopg.connect("postgresql://...", row_factory=dict_row) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT id, name FROM users")
        for row in cur:
            print(row["name"])

Row factories can also be set per cursor, giving you flexibility when you need tuple access in one place and dictionary access in another. This makes code more readable and reduces positional indexing errors.