使用SQLAlchemy执行原生SQL查询报错，如何获取查询结果？

问题代码 #

import sqlalchemy.pool as pool
import psycopg2
from dotenv import load_dotenv
import os
load_dotenv()


def getconn():
    c = psycopg2.connect(user=os.getenv('POSTGRES_USER'), password = os.getenv('POSTGRES_PASSWORD'),  host=os.getenv('POSTGRES_HOST'), port=os.getenv('DATABASE_PORT'), dbname=os.getenv('POSTGRES_DB'))
    return c


mypool = pool.QueuePool(getconn, max_overflow=10, pool_size=5)


def get_connection():
    conn = mypool.connect()
    return conn.cursor()

# use it
db = get_connection()
print(db)
output = db.execute("select 2+2 as sum")
print(output.fetchone())

错误信息 #

<cursor object at 0x0000021B18795C40; closed: 0>
Traceback (most recent call last):
  File "C:\Users\gourav.kumar\code\genai_ozone_engine\genai_ozone_engine\app\database\database.py", line 25, in <module>
    print(output.fetchone())
AttributeError: 'NoneType' object has no attribute 'fetchone'

问题原因 #

psycopg2的cursor.execute()方法本身返回None，查询结果直接存储在cursor对象中，并非execute方法的返回值。你将execute的返回值赋值给output，自然得到None，调用fetchone()就会触发错误。

修正方案 #

方案1：基于现有连接池代码修正 #

直接使用cursor对象获取结果，同时注意资源回收：

import sqlalchemy.pool as pool
import psycopg2
from dotenv import load_dotenv
import os
load_dotenv()


def getconn():
    c = psycopg2.connect(
        user=os.getenv('POSTGRES_USER'),
        password=os.getenv('POSTGRES_PASSWORD'),
        host=os.getenv('POSTGRES_HOST'),
        port=os.getenv('DATABASE_PORT'),
        dbname=os.getenv('POSTGRES_DB')
    )
    return c


mypool = pool.QueuePool(getconn, max_overflow=10, pool_size=5)


def get_connection():
    conn = mypool.connect()
    return conn, conn.cursor()

# 使用示例
conn, db = get_connection()
try:
    # 执行查询，无需接收返回值
    db.execute("select 2+2 as sum")
    # 直接通过cursor对象获取结果
    result = db.fetchone()
    print(result)  # 输出: (4,)
finally:
    # 关闭cursor并将连接归还到池
    db.close()
    conn.close()

如果需要字典格式的结果，可以修改cursor的创建方式：

# 在get_connection函数中修改cursor创建
return conn, conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
# 此时fetchone()会返回: {'sum': 4}

方案2：使用SQLAlchemy 2.x原生方式执行原生SQL #

SQLAlchemy本身提供了更简洁的原生SQL执行方式，无需直接操作psycopg2：

from sqlalchemy import create_engine, text
from dotenv import load_dotenv
import os

load_dotenv()

# 创建SQLAlchemy Engine
engine = create_engine(
    f"postgresql+psycopg2://{os.getenv('POSTGRES_USER')}:{os.getenv('POSTGRES_PASSWORD')}@{os.getenv('POSTGRES_HOST')}:{os.getenv('DATABASE_PORT')}/{os.getenv('POSTGRES_DB')}"
)

# 执行查询
with engine.connect() as conn:
    result = conn.execute(text("select 2+2 as sum"))
    print(result.fetchone())  # 输出: (4,)
    # 若为写操作，需要提交事务
    # conn.commit()

内容的提问来源于stack exchange，提问作者gourav kumar