如何使用 SQLAlchemy 在多对多结构上使用 Python 列表(必须具有所有或必须具有至少一个元素)进行查询?

How to do a query using Python list (must have all OR must have at least one element) on a many to many structure with SQLAlchemy?

我将创建一个不具体的结构,以便问题变得更容易理解。考虑存储书籍信息的数据库结构:

from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow

db = SQLAlchemy()

#Base class to create some good pratices fields
class Base(db.Model):
    __abstract__  = True

    id            = db.Column(db.Integer, primary_key=True)
    date_created  = db.Column(db.DateTime,  default=db.func.current_timestamp())
    date_modified = db.Column(db.DateTime,  default=db.func.current_timestamp(),
                                           onupdate=db.func.current_timestamp())
class Book(Base):
    name = db.Column(db.String(255), nullable=False)
    pages = db.Column(db.Integer, nullable=True)
    tags = db.relationship('BooksTags', back_populates="book")

class BooksTags(db.Model):
    __tablename__ = 'book_tag'
    tag_id = db.Column(db.Integer, db.ForeignKey('book.id'), primary_key=True)
    book_id = db.Column(db.Integer, db.ForeignKey('tag.id'), primary_key=True)
    book = db.relationship('Book', back_populates='tags')
    tag = db.relationship('Tag', back_populates='books')

class Tags(Base):
    name = db.Column(db.String(255), nullable=False)
    books = db.relationship('BooksTags', back_populates="tag")

使用这段代码,我有一个“多对多”结构,对我来说效果很好,但是当我尝试基于标签创建查询时,我遇到了困难,例如:

我一直在尝试使用一些聚合函数,例如 array_agg 和 group_concat,但我总是 returns 不同的错误消息,我不知道该怎么做。

纯 SQL 我会这样查询:

SELECT 
  group_concat(tag.name) AS tags, 
  book.name as book_name
FROM 
  book 
  INNER JOIN book_tag ON book.id = book_tag.book_id 
  INNER JOIN tag ON tag.id = book_tag.tag_id 
GROUP BY 
  book.id

但是 IDK 如何过滤这个查询。

好的,在阅读并询问了一些朋友的帮助后,我已经设法做到了 SQL:

在“至少一个”标签的情况下:

SELECT 
  group_concat(tag.name) AS tags, 
  book.name as book_name
FROM 
  book 
  INNER JOIN book_tag ON book.id = book_tag.book_id 
  INNER JOIN tag ON tag.id = book_tag.tag_id AND tag.name in (<insert tag list>)
GROUP BY 
  book.id

在“必须有所有”标签的情况下:

SELECT 
  group_concat(tag.name) AS tags, 
  book.name as book_name
FROM 
  book 
  INNER JOIN book_tag ON book.id = book_tag.book_id 
  INNER JOIN tag ON tag.id = book_tag.tag_id AND tag.name in (<insert tag list>)
GROUP BY 
  book.id
WHERE
  COUNT(tag.name) > <sizeof tag array>

但是我还是不知道如何用SQL炼金术。

您应该可以按如下方式过滤您的图书:

filtered = Book.query().filter(Book.tags.tag.name == 'tag_name')

根据您的标准,您可以制作一个循环来根据您的需要进行塑造。

您可以使用

在原始过滤器之上设置额外的过滤器
filtered.filter()

经过我的编辑,问题更容易解决,所以做一个 select 所有至少有一个数组标签的书:

tags_id = [1,2,3]
books_query = Book.query.join(BooksTags, Tag)
books_query = books_query.filter(BooksTags.tag_id.in_(tags_id))
books_query = books_query.group_by(Book)

这将return像这样

SELECT 
  book.id as book_id,
  book.name as book_name,
  book.pages as book_pages
FROM
  book
  INNER JOIN book_tag ON book.id = book_tag.book_id 
  INNER JOIN tag ON tag.id = book_tag.tag_id 
WHERE 
  book_tag.tag_id IN (
        1, 
        2,
        3
  ) 
GROUP BY 
  book.id,
  book.name,
  book.pages

然后做一个 select 所有具有数组所有标签的书:

tags_id = [1,2,3]
books_query = Book.query.join(BooksTags, Tag)
books_query = books_query.filter(BooksTags.tag_id.in_(tags_id))
books_query = books_query.group_by(Book)
books_query = books_query.having(func.count(Book.id) >= len(tags_id))

这将return像这样

SELECT 
  book.id as book_id,
  book.name as book_name,
  book.pages as book_pages
FROM
  book
  INNER JOIN book_tag ON book.id = book_tag.book_id 
  INNER JOIN tag ON tag.id = book_tag.tag_id 
WHERE 
  book_tag.tag_id IN (
        1, 
        2,
        3
  ) 
GROUP BY 
  book.id,
  book.name,
  book.pages
HAVING 
  count(book.id) >= 3