使用 SQLAlchemy 执行不指定字段名的 SQL 查询

SQLAlchemy 提供了多种方式来执行原始 SQL 查询，包括不指定字段名的情况。以下是几种常见方法：

1. 使用 text() 和 execute()

from sqlalchemy import create_engine, text

# 创建引擎
engine = create_engine('数据库连接字符串')

# 执行不指定字段名的查询
with engine.connect() as connection:
    result = connection.execute(text("SELECT * FROM users WHERE age > :age"), {"age": 25})

    # 获取结果
    for row in result:
        print(row)  # 返回的是元组
        # 或者
        print(row[0], row[1])  # 通过索引访问

2. 使用 session.execute()

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

# 执行查询
result = session.execute(text("SELECT * FROM products WHERE price > :price"), {"price": 100})

# 处理结果
for row in result:
    print(row)  # 元组形式
    print(f"ID: {row[0]}, Name: {row[1]}")

3. 使用 engine.execute() (较旧的方法)

result = engine.execute("SELECT * FROM orders WHERE status = 'shipped'")

for row in result:
    print(row)  # 返回的是 RowProxy 对象，可以像字典或元组一样访问

4. 将结果转换为字典

如果你想要更方便地访问字段（但仍然不知道字段名）：

with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM customers"))

    # 获取字段名
    columns = result.keys()

    for row in result:
        # 将行转换为字典
        row_dict = dict(zip(columns, row))
        print(row_dict)

注意事项

1. 不指定字段名虽然方便，但会使代码更脆弱，因为表结构变化可能破坏查询
2. 在生产环境中，建议明确指定需要的字段
3. 对于复杂查询，考虑使用 SQLAlchemy Core 或 ORM 来构建查询
4. 注意 SQL 注入风险，始终使用参数化查询（如上面的 :param 语法）

性能考虑

对于大型结果集，可以考虑使用：

# 使用 fetchmany 分批获取
result = connection.execute(text("SELECT * FROM large_table"))
while True:
    chunk = result.fetchmany(1000)  # 每次获取1000行
    if not chunk:
        break
    for row in chunk:
        process_row(row)

希望这些方法能帮助你在 SQLAlchemy 中执行不指定字段名的 SQL 查询。