在 sqlalchemy 中过滤 returns a table 的 PostgreSQL 函数

Question

我正在尝试解决以下两个问题：

构建一个 ORM 模型以在 SQLAlchemy 中使用，table 从 PostgreSQL 中的用户定义函数生成
过滤说 table（无论是否在 ORM 模型中）

我定义了一个 returns 一个 table 调用 transformed 的函数。它需要 user_id 作为输入和 returns 3 列。看起来像下面这样（这是非常简化的，查询显然要复杂一些）。

CREATE OR REPLACE FUNCTION transformed (user_id integer)
    RETURNS TABLE (
        id integer,
        user varchar,
        description varchar,
    )
    AS $body$
SELECT
    id,
    user,
    description
FROM table
WHERE id = 

$body$
LANGUAGE sql;

def build_query(user_id):
    base_query: Query = select(
        Column("id", Integer),
        Column("user", String),
        Column("description", String),
    ).select_from(func.transformed)_data(user_id))
    return base_query

获取三列没问题。挑战在于对它们进行过滤和排序。

query = build_query(user_id).filter_by(description = "Foo")

此returns以下错误：sqlalchemy.exc.InvalidRequestError: Entity namespace for "transformed(:transformed_1) has no property "description".

我很想知道如何 a) 正确过滤或 b) 将我定义的纯列转换为某种 ORM 模型，然后我可以使用它来过滤 ORM 模型的属性。

Answer 1

有一种可能的解决方案，但它与通过原始文本简单地构建查询非常相似：

query = build_query(user_id).where(text("id = 4"))

不太喜欢，因为动态扩展它并不容易。

Answer 2

SQLAlchemy 非常棒。您可以使用 DDLEvents and then call the function using func and define the output with table_valued 定义函数，标量有不同的方法，但它在同一文档中。


import sys
from sqlalchemy import (
    create_engine,
    Integer,
    String,
    event,
    column,
)
from sqlalchemy.schema import (
    Column,
    DDL,
)
from sqlalchemy.sql import select, func
from sqlalchemy.orm import declarative_base, Session


Base = declarative_base()


username, password, db = sys.argv[1:4]


engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=True)


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(8), index=True)


event.listen(
    User.__table__,
    "after_create",
    DDL("""CREATE OR REPLACE FUNCTION transformed (int) RETURNS TABLE (id integer, username varchar)
    AS $$ SELECT id, name AS username FROM users WHERE id =  $$
    LANGUAGE sql;
"""))


Base.metadata.create_all(engine)



with Session(engine) as session, session.begin():
    for name in ['User 1', 'User 2', 'User 3']:
        session.add(User(name=name))

with Session(engine) as session, session.begin():
    # Get a single row with id 1
    fn = func.transformed(1).table_valued(
        column("id", Integer),
        column("username", String))

    # This filtering is redundant because we selected
    # a user by its id but it demonstrates referencing the column
    # we defined above.
    q = select(fn).where(fn.c.username.like('User %'))
    for id, user in session.execute(q).all():
        print(id, user)

这会打印出来

1 User 1

实际查询如下所示，来自 echo=True:

SELECT anon_1.id, anon_1.username 
FROM transformed(%(transformed_1)s) AS anon_1 
WHERE anon_1.username LIKE %(username_1)s

在 sqlalchemy 中过滤 returns a table 的 PostgreSQL 函数

Filtering a PostgreSQL function that returns a table in sqlalchemy

python

sqlalchemy