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

Question

我见过this 和this 的问题，但它们都是针对旧的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'>

附：如果可能的话，最好不要从sqlalchemy 导入select，而是使用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)

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

如果你想查询一个ORM模型类的属性，语法是similar，但是你直接访问列

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())