如何分配自定义 SQL 查询，该查询返回行集合作为 ORM 模型的属性

Posted 2023-03-16

【中文标题】如何分配自定义 SQL 查询，该查询返回行集合作为 ORM 模型的属性

【英文标题】：How to assign a custom SQL query which returns a collection of Rows as an attribute of an ORM model

【发布时间】：2022-01-22 11:35:49

【问题描述】：

例如，假设我有三个模型：Book、Author 和 BookAuthor，其中一本书可以有很多作者，而一个作者可以有很多书。

class BookAuthor(Base):
    __tablename__ = 'book_authors'
    author_id = Column(ForeignKey('authors.id'), primary_key=True)
    book_id = Column(ForeignKey('books.id'), primary_key=True)
    blurb = Column(String(50))

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)

我想创建一个Book 的authors 属性，它返回该书的每个作者和关于每个作者的相应简介。像这样的

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)

    @authors.expression
    def authors(cls):
        strSQL = "my custom SQL query"
        return execute(strSQL)

---

演示

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, Session

# Make the engine
engine = create_engine("sqlite+pysqlite:///:memory:", future=True, echo=False)

# Make the DeclarativeMeta
Base = declarative_base()

class BookAuthor(Base):
    __tablename__ = 'book_authors'
    author_id = Column(ForeignKey('authors.id'), primary_key=True)
    book_id = Column(ForeignKey('books.id'), primary_key=True)
    blurb = Column(String(50))

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)

# Create the tables in the database
Base.metadata.create_all(engine)

# Make data
with Session(bind=engine) as session:

    # add parents
    a1 = Author()
    session.add(a1)

    a2 = Author()
    session.add(a2)

    session.commit()

    # add children
    b1 = Book()
    session.add(b1)

    b2 = Book()
    session.add(b2)

    session.commit()

    # map books to authors
    ba1 = BookAuthor(author_id=a1.id, book_id=b1.id, blurb='foo')
    ba2 = BookAuthor(author_id=a1.id, book_id=b2.id, blurb='bar')
    ba3 = BookAuthor(author_id=a2.id, book_id=b2.id, blurb='baz')

    session.add(ba1)
    session.add(ba2)
    session.add(ba3)

    session.commit()

# Get the authors for book with id 2
with Session(bind=engine) as session:
    s = """
        SELECT foo.* FROM (
            SELECT 
                authors.*, 
                book_authors.blurb, 
                book_authors.book_id 
            FROM authors INNER JOIN book_authors ON authors.id = book_authors.author_id
        ) AS foo
        INNER JOIN books ON foo.book_id = books.id
        WHERE books.id = :bookid
        """
    result = session.execute(s, params='bookid':2).fetchall()
    print(result)

看到最后那个半讨厌的查询了吗？它成功返回了第 2 本书的作者，包括有关每个作者的相应简介。我想为我的Book 模型创建一个.authors 属性来执行这个查询。

【参考方案1】：

想通了。关键是要使用plain descriptor 和object_session()

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)

    @property
    def authors(self):
        s = """
                SELECT foo.* FROM (
                    SELECT
                        authors.*,
                        book_authors.blurb,
                        book_authors.book_id
                    FROM authors INNER JOIN book_authors ON authors.id = book_authors.author_id
                ) AS foo
                INNER JOIN books ON foo.book_id = books.id
                WHERE books.id = :bookid
                """
        result = object_session(self).execute(s, params='bookid': self.id).fetchall()
        return result