Sqlalchemy 仅选择一列（语法 > 1.4！）

Question

我看过这个和这个问题，但它们都是针对旧 Sqlalchemy 版本的。我在查询中使用以下语法：

get_user_st = users.select().where(users.c.login == user.phone_number)
connection.execute(statement=get_user_st).fetchone()

这是我选择的phone_number。如何选择整列？

我已经尝试过的错误语法：

str(users.select(users.c.login))
'SELECT users.id, users.phone_number, users.email, users.login, users.full_name, users.position, users.hashed_password, users.role, users.created_datetime, users.last_update, users.owner_id \nFROM users \nWHERE users.login'

str(users.c.login.select())
Traceback (most recent call last):
  File "/snap/pycharm-community/238/plugins/python-ce/helpers/pydev/_pydevd_bundle/pydevd_exec2.py", line 3, in Exec
    exec(exp, global_vars, local_vars)
  File "<input>", line 1, in <module>
  File "/home/david/PycharmProjects/na_svyazi/venv/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 818, in __getattr__
    util.raise_(
  File "/home/david/PycharmProjects/na_svyazi/venv/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
AttributeError: Neither 'Column' object nor 'Comparator' object has an attribute 'select'

我试图在 Sqlalchemy 文档中找到这个案例，但失败了。

获取数据库：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError


engine_async = create_async_engine(url=config.SQLALCHEMY_ASYNCPG_URL, echo=False, future=True)  # Future -use style 2.0
session_async = sessionmaker(bind=engine_async, autoflush=True, class_=AsyncSession)

async def get_db():
    session = session_async()
    try:
        yield session
        await session.commit()
    except SQLAlchemyError as ex:
        await session.rollback()
        raise ex
    finally:
        await session.close()

我的会话类型是：

type(postgres_session)
<class 'sqlalchemy.orm.session.AsyncSession'>

PS 如果可能的话，最好不要select从中导入sqlalchemy，而是使用 Table 对象（users.c或仅使用users）

Answer 1

如果您使用的是 SQLAlchemy 核心。select不要使用表实例的方法，而是使用select函数（docs），或者如果需要使用表的方法，请使用select.with_only_columns。

import sqlalchemy as sa


engine = sa.create_engine('postgresql:///test', echo=True, future=True)
Users = sa.Table('users', sa.MetaData(), autoload_with=engine)

with engine.begin() as conn:
    q = Users.select().with_only_columns(Users.c.id, Users.c.name)
    res = conn.execute(q)
    for row in res:
        print(row)

请注意，此核心行为对于 SQLAlchemy 1.4 来说并不新鲜，尽管 的语法select略有变化。

如果要查询 ORM 模型类的属性，语法类似，但是直接访问列

q = sa.select(User.name)
result = session.execute(q)

如果您想使用异步驱动程序，代码可能如下所示：

import asyncio

import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async def async_main():
    engine = create_async_engine(
        "postgresql+asyncpg:///test", echo=True, future=True
    )

    async with engine.connect():

        Session = orm.sessionmaker(engine, class_=AsyncSession)
        session = Session()
        # Users is the table from the earlier example
        result = await session.execute(
            Users.select.with_only_columns(Users.c.name)
        )
        print(result.fetchall())

        await session.close()
    await engine.dispose()


asyncio.run(async_main())