Fix psycopg

Author: jy-tan

drift/instrumentation/psycopg/instrumentation.py

@@ -1888,35 +1888,41 @@ def _finalize_executemany_returning_span(
                 }
                 span.set_status(Status(OTelStatusCode.ERROR, str(error)))
             else:
-                # Iterate through cursor.results() to capture all result sets
+                # Iterate through result sets using nextset() (compatible with psycopg 3.2+)
+                # Note: results() requires psycopg 3.3+, so we use nextset() for broader compatibility
                 result_sets = []
                 all_rows_collected = []  # For re-populating cursor
 
                 try:
-                    # cursor.results() yields the cursor itself for each result set
-                    for result_cursor in cursor.results():
+
+                    def capture_current_result_set():
+                        """Capture the current result set's description and rows.
+
+                        Extracts column metadata from cursor.description, fetches all rows,
+                        normalizes row data (handling dict, namedtuple, and tuple formats),
+                        and serializes for storage. Appends raw rows to all_rows_collected
+                        for later cursor re-population.
+                        """
                         result_set_data = {}
 
-                        # Capture description for this result set
-                        if hasattr(result_cursor, "description") and result_cursor.description:
+                        if hasattr(cursor, "description") and cursor.description:
                             description = [
                                 {
                                     "name": desc[0] if hasattr(desc, "__getitem__") else desc.name,
                                     "type_code": desc[1]
                                     if hasattr(desc, "__getitem__") and len(desc) > 1
                                     else getattr(desc, "type_code", None),
                                 }
-                                for desc in result_cursor.description
+                                for desc in cursor.description
                             ]
                             result_set_data["description"] = description
                             column_names = [d["name"] for d in description]
                         else:
                             description = None
                             column_names = None
 
-                        # Fetch all rows for this result set
                         rows = []
-                        raw_rows = result_cursor.fetchall()
+                        raw_rows = cursor.fetchall()
                         all_rows_collected.append(raw_rows)
 
                         for row in raw_rows:
@@ -1933,15 +1939,18 @@ def _finalize_executemany_returning_span(
                             else:
                                 rows.append(list(row))
 
-                        result_set_data["rowcount"] = (
-                            result_cursor.rowcount if hasattr(result_cursor, "rowcount") else len(rows)
-                        )
+                        result_set_data["rowcount"] = cursor.rowcount if hasattr(cursor, "rowcount") else len(rows)
                         result_set_data["rows"] = [[serialize_value(col) for col in row] for row in rows]
 
-                        result_sets.append(result_set_data)
+                        return result_set_data
+
+                    result_sets.append(capture_current_result_set())
+
+                    while cursor.nextset():
+                        result_sets.append(capture_current_result_set())
 
                 except Exception as results_error:
-                    logger.debug(f"Could not iterate results(): {results_error}")
+                    logger.debug(f"Could not iterate result sets: {results_error}")
                     # Fallback: treat as single result set
                     result_sets = []