Psycopg2未抛出‘无结果可获取’错误的技术求助

Let's break down why you're not seeing the expected "no results to fetch" error from Psycopg2 and walk through fixes to resolve this issue.

Why the Error Isn't Triggering #

Your current code has two key issues that prevent the Psycopg2 error from firing:

1. Psycopg2's Error Logic Doesn't Work That Way
   Psycopg2 only throws ProgrammingError: no results to fetch if you try to fetch results from a cursor that hasn't executed a query at all. For:

   • A SELECT query with no matching rows: fetchall() returns an empty list (no error thrown).
   • Non-query statements like INSERT/UPDATE/DELETE: fetchall() also returns an empty list (since these don't produce a result set).

   In your code, you immediately access cur.fetchall()[0]—so when there are no results, you'll get a Python native IndexError: list index out of range instead of the Psycopg2 error you're expecting.

2. No Distinction Between Query and Non-Query Statements
   Your execute_statement function treats all SQL the same, but queries (SELECT) and data manipulation statements (INSERT/UPDATE) need different handling. You shouldn't try to fetch results from non-query operations at all.

Fixes to Implement #

1. Split Functions for Queries vs. Non-Queries #

The cleanest approach is to separate logic for operations that return results and those that don't:

import psycopg2 as psdb

def execute_query(stmt, params=None):
    """Run a SELECT query and return the first result row (or raise an error if empty)"""
    conn = psdb.connect(dbname='db', user='user', host='localhost', password='password')
    cur = conn.cursor()
    try:
        cur.execute(stmt, params or ())
        rows = cur.fetchall()
        if not rows:
            # Manually raise the Psycopg2-style error for consistency
            raise psdb.ProgrammingError("no results to fetch")
        return rows[0]
    except psdb.Error as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

def execute_non_query(stmt, params=None):
    """Run INSERT/UPDATE/DELETE and return the number of affected rows"""
    conn = psdb.connect(dbname='db', user='user', host='localhost', password='password')
    cur = conn.cursor()
    try:
        cur.execute(stmt, params or ())
        conn.commit()
        return cur.rowcount
    except psdb.Error as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

2. Update the Original Function with Result Checks #

If you want to keep a single function, use cur.description to detect if the statement produced a result set, then handle empty results explicitly:

import psycopg2 as psdb

def execute_statement(stmt, params=None): 
    conn = psdb.connect(dbname='db', user='user', host='localhost', password='password') 
    cur = conn.cursor() 
    try:
        cur.execute(stmt, params or ())
        # Check if we have a result set (i.e., this was a SELECT query)
        if cur.description is not None:
            rows = cur.fetchall()
            if not rows:
                raise psdb.ProgrammingError("no results to fetch")
            return rows[0]
        else:
            # Non-query statement: commit and return affected rows
            conn.commit()
            return cur.rowcount
    except psdb.Error as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

3. Adjust Error Handling in get_product_details #

Update your caller function to catch the specific errors and handle missing products properly:

def get_product_details(request): 
    """Retrieve all information for a certain product, -> returns id, name, description and price""" 
    faultstring = None 
    try: 
        # Use parameterized queries to avoid SQL injection!
        row = execute_statement(
            """SELECT id, name, description, price FROM products WHERE id = %s""",
            (request.product_id,)
        )
        return {"id": row[0], "name": row[1], "description": row[2], "price": row[3]}
    except psdb.ProgrammingError as e:
        if "no results to fetch" in str(e):
            faultstring = "Requested product not found"
        else:
            faultstring = f"Database query error: {str(e)}"
    except Exception as e:
        faultstring = f"Unexpected error: {str(e)}"
    
    if faultstring:
        return {"error": faultstring}

Critical Additional Notes #

• Avoid SQL Injection: Always use parameterized queries (the params argument in execute()) instead of string concatenation to build SQL statements.
• Transaction Safety: Always commit non-query statements and roll back on errors to prevent data inconsistencies.
• Resource Cleanup: Using try...finally ensures database connections are closed even if an error occurs, preventing resource leaks.

内容的提问来源于stack exchange，提问作者Nathan Ruth