如何从SQLAlchemy异步查询结果构建Polars DataFrame？

动态提取列名并构建Polars LazyFrame #

你当前的查询返回ORM实体实例，所以result.keys()会显示模型类名而非列名。要实现无硬编码列名构建懒加载DataFrame，可按以下两种方式调整：

方法一：直接查询表列（推荐，效率更高） #

修改查询语句，直接指定查询User表的所有列，这样返回的结果集包含原始列数据和真实列名：

import polars as pl
from sqlalchemy import select

async with EngineContext(uri=URI) as engine:
    session = async_sessionmaker(bind=engine, expire_on_commit=True)()
    async with session.begin():
        # 直接查询表的所有列，而非ORM实例
        stmt = select(*User.__table__.columns).order_by(User.login_date.desc()).limit(10)
        result = await session.execute(stmt)
        
        # 获取列名和行数据
        columns = result.keys()
        rows = result.fetchall()
        
        # 构建Polars懒加载DataFrame
        lf = pl.LazyFrame(data=rows, schema=columns)
        # 按需执行懒操作后，用.collect()触发计算
        final_df = lf.collect()

方法二：从ORM实例动态提取属性（适合必须用ORM实例的场景） #

如果要保留ORM实例查询，可从模型元数据获取列名，再遍历实例提取对应属性：

import polars as pl
from sqlalchemy import select

async with EngineContext(uri=URI) as engine:
    session = async_sessionmaker(bind=engine, expire_on_commit=True)()
    async with session.begin():
        stmt = select(User).order_by(User.login_date.desc()).limit(10)
        result = await session.execute(stmt)
        
        # 从模型元数据获取所有列名
        columns = [col.name for col in User.__table__.columns]
        # 遍历实例提取列对应属性值
        rows = [{col: getattr(user, col) for col in columns} for user in result.scalars().all()]
        
        # 构建LazyFrame
        lf = pl.LazyFrame(rows)

将SQLAlchemy会话标记为只读 #

有三种实用方式避免无意义的写入操作：

方式一：会话级别只读控制 #

在创建会话或启动事务时指定只读选项：

# 方式1：创建会话时默认开启只读
async with EngineContext(uri=URI) as engine:
    session = async_sessionmaker(
        bind=engine,
        expire_on_commit=True,
        execute_options={"read_only": True}
    )()
    async with session.begin():
        # 该会话内无法执行add/update/delete等写入操作
        stmt = select(*User.__table__.columns).order_by(User.login_date.desc()).limit(10)
        result = await session.execute(stmt)

# 方式2：仅在当前事务内设置只读
async with session.begin(read_only=True):
    # 此事务块内为只读模式

方式二：引擎级别全局只读 #

如果所有基于该引擎的操作都需要只读，可在创建引擎时配置：

from sqlalchemy import create_async_engine

engine = create_async_engine(
    URI,
    execution_options={"read_only": True}
)

方式三：数据库用户权限限制（最安全） #

直接在数据库层面创建只读用户，仅赋予User表的SELECT权限，即使代码存在写入逻辑，数据库也会直接拒绝执行，从根源避免数据损坏。

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