如何在 SQLAlchemy 中设置 M2M 混合计数 属性？
How to set a M2M hybrid count property in SQLAlchemy?
我有两个由 M2M 关系绑定的表。书和作家，作家可以有很多书，书可以有很多作家。
我想在书籍和作家上都有一个 count 属性，这样我就可以将它们排序，例如，写书最多的作家。
# many to many association table
book_writer_association_table = Table('book_writer_association',Base.metadata,
Column('book_id',ForeignKey('book.id'), primary_key=True),
Column('Writer',ForeignKey('writer.id'), primary_key=True)
)
class Book(Base):
__tablename__ = 'base'
id = Column(Integer, primary_key=True)
name = Column(String)
writers = relationship(Writer,secondary=book_writer_association_table,back_populates="books")
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship(Book,secondery=book_writer_association_table,back_populates="writers")
@hybrid_property
def book_count(self):
return len(self.books)
@book_count.expression
def book_count(cls):
#what goes here?
我尝试了各种方法，例如详细 here:
class Foo(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
bar_id = Column(Integer, ForeignKey('bar.id'))
bar = relationship('Bar')
class Bar(Base):
__tablename__ = 'bar'
id = Column(Integer, primary_key=True)
@hybrid_property
def foo_count(self):
return object_session(self).query(Foo).filter(Foo.bar==self).count()
@foo_count.expression
def foo_count(cls):
return select([func.count(Foo.id)]).where(Foo.bar_id == cls.id).label('foo_count')
但是，在这个例子中，只有两个表，我不确定如何在这里实现更复杂的连接。另一位用户建议使用 column_property 但我 运行 遇到了完全相同的问题。我不确定如何进一步向连接添加表。
您可以自定义想法，从 here 到 M2M 案例。为此，您应该在 hybrid_property 中提及 association_table 而不是 Book table。因此，您消除了与 Book table 的连接，并将您的案例简化为一对多关系。
我想到了这个解决方案。
from typing import List
from sqlalchemy import Column, ForeignKey, Integer, String, select, func, create_engine, Table
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, object_session, sessionmaker, Session
# Declare models
@as_declarative()
class Base:
pass
book_writer_association_table = Table('book_writer_association',Base.metadata,
Column('book_id',ForeignKey('book.id'), primary_key=True),
Column('writer_id',ForeignKey('writer.id'), primary_key=True)
)
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True)
name = Column(String)
writers = relationship("Writer", secondary=book_writer_association_table, back_populates="books")
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship("Book", secondary=book_writer_association_table, back_populates="writers")
@hybrid_property
def book_count(self):
return object_session(self).query(book_writer_association_table).filter(book_writer_association_table.c.writer_id == self.id).count()
@book_count.expression
def book_count(cls):
return select([func.count(book_writer_association_table.c.book_id)]).where(book_writer_association_table.c.writer_id == cls.id).label('book_count')
# Load DB schema
engine = create_engine('sqlite:///sqlite3.db')
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=True, bind=engine)
db: Session = SessionLocal()
# Creating test instances
b1 = Book(name="Book 1")
b2 = Book(name="Book 2")
db.add(b1)
db.add(b2)
w1 = Writer(name="Writer 1")
w2 = Writer(name="Writer 2")
db.add(w1)
db.add(w2)
b1.writers.append(w1)
b1.writers.append(w2)
b2.writers.append(w1)
query = db.query(Writer, Writer.book_count)
print(str(query)) # checking query
print()
writers: List[Writer] = query.all() # testing query
for writer, book_count in writers:
print(f"{writer.name}: {book_count}")
结果：
> Writer 1: 2
> Writer 2: 1
I'm unsure how to further add tables to the join.
SQL 从这里 db.query(Writer, Writer.book_count) 看起来很干净，没有任何连接。所以，我认为你应该不会对后续加入有任何问题。
> SELECT writer.id AS writer_id, writer.name AS writer_name, (SELECT count(book_writer_association.book_id) AS count_1
> FROM book_writer_association
> WHERE book_writer_association.writer_id = writer.id) AS book_count
> FROM writer
编辑：如果您需要加入Book table以提供额外的过滤，您可以这样做。这里我筛选了价格低于150
的书
from typing import List
from sqlalchemy import Column, ForeignKey, Integer, String, select, func, create_engine, Table
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, object_session, sessionmaker, Session
# Declare models
@as_declarative()
class Base:
pass
book_writer_association_table = Table('book_writer_association',Base.metadata,
Column('book_id',ForeignKey('book.id'), primary_key=True),
Column('writer_id',ForeignKey('writer.id'), primary_key=True)
)
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True)
name = Column(String)
price = Column(Integer)
writers = relationship("Writer", secondary=book_writer_association_table, back_populates="books")
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship("Book", secondary=book_writer_association_table, back_populates="writers")
@hybrid_property
def book_count(self):
return (
object_session(self)
.query(book_writer_association_table)
.join(Book, Book.id == book_writer_association_table.c.book_id)
.filter(book_writer_association_table.c.writer_id == self.id)
.filter(Book.price > 150)
.count()
)
@book_count.expression
def book_count(cls):
# return select([func.count(book_writer_association_table.c.book_id)]).where(book_writer_association_table.c.writer_id == cls.id).label('book_count')
#
return (
select([func.count(book_writer_association_table.c.book_id)])
.join(Book, Book.id == book_writer_association_table.c.book_id)
.where(book_writer_association_table.c.writer_id == cls.id)
.filter(Book.price > 150)
.label('book_count')
)
# Load DB schema
engine = create_engine('sqlite:///sqlite3.db')
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=True, bind=engine)
db: Session = SessionLocal()
# Creating test instances
b1 = Book(name="Book 1", price=100)
b2 = Book(name="Book 2", price=200)
db.add(b1)
db.add(b2)
w1 = Writer(name="Writer 1")
w2 = Writer(name="Writer 2")
db.add(w1)
db.add(w2)
b1.writers.append(w1)
b1.writers.append(w2)
b2.writers.append(w1)
query = db.query(Writer, Writer.book_count)
print(str(query)) # checking query
print()
writers: List[Writer] = query.all() # testing query
for writer, book_count in writers:
print(f"{writer.name}: {book_count}")
查询：
SELECT writer.id AS writer_id,
writer.name AS writer_name,
(SELECT count(book_writer_association.book_id) AS count_1
FROM book_writer_association
JOIN book ON book.id = book_writer_association.book_id
WHERE book_writer_association.writer_id = writer.id
AND book.price > ?) AS book_count
FROM writer
我有两个由 M2M 关系绑定的表。书和作家，作家可以有很多书，书可以有很多作家。
我想在书籍和作家上都有一个 count 属性，这样我就可以将它们排序，例如，写书最多的作家。
# many to many association table
book_writer_association_table = Table('book_writer_association',Base.metadata,
Column('book_id',ForeignKey('book.id'), primary_key=True),
Column('Writer',ForeignKey('writer.id'), primary_key=True)
)
class Book(Base):
__tablename__ = 'base'
id = Column(Integer, primary_key=True)
name = Column(String)
writers = relationship(Writer,secondary=book_writer_association_table,back_populates="books")
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship(Book,secondery=book_writer_association_table,back_populates="writers")
@hybrid_property
def book_count(self):
return len(self.books)
@book_count.expression
def book_count(cls):
#what goes here?
我尝试了各种方法，例如详细 here:
class Foo(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
bar_id = Column(Integer, ForeignKey('bar.id'))
bar = relationship('Bar')
class Bar(Base):
__tablename__ = 'bar'
id = Column(Integer, primary_key=True)
@hybrid_property
def foo_count(self):
return object_session(self).query(Foo).filter(Foo.bar==self).count()
@foo_count.expression
def foo_count(cls):
return select([func.count(Foo.id)]).where(Foo.bar_id == cls.id).label('foo_count')
但是，在这个例子中，只有两个表，我不确定如何在这里实现更复杂的连接。另一位用户建议使用 column_property 但我 运行 遇到了完全相同的问题。我不确定如何进一步向连接添加表。
您可以自定义想法，从 here 到 M2M 案例。为此，您应该在 hybrid_property 中提及 association_table 而不是 Book table。因此，您消除了与 Book table 的连接，并将您的案例简化为一对多关系。
我想到了这个解决方案。
from typing import List
from sqlalchemy import Column, ForeignKey, Integer, String, select, func, create_engine, Table
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, object_session, sessionmaker, Session
# Declare models
@as_declarative()
class Base:
pass
book_writer_association_table = Table('book_writer_association',Base.metadata,
Column('book_id',ForeignKey('book.id'), primary_key=True),
Column('writer_id',ForeignKey('writer.id'), primary_key=True)
)
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True)
name = Column(String)
writers = relationship("Writer", secondary=book_writer_association_table, back_populates="books")
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship("Book", secondary=book_writer_association_table, back_populates="writers")
@hybrid_property
def book_count(self):
return object_session(self).query(book_writer_association_table).filter(book_writer_association_table.c.writer_id == self.id).count()
@book_count.expression
def book_count(cls):
return select([func.count(book_writer_association_table.c.book_id)]).where(book_writer_association_table.c.writer_id == cls.id).label('book_count')
# Load DB schema
engine = create_engine('sqlite:///sqlite3.db')
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=True, bind=engine)
db: Session = SessionLocal()
# Creating test instances
b1 = Book(name="Book 1")
b2 = Book(name="Book 2")
db.add(b1)
db.add(b2)
w1 = Writer(name="Writer 1")
w2 = Writer(name="Writer 2")
db.add(w1)
db.add(w2)
b1.writers.append(w1)
b1.writers.append(w2)
b2.writers.append(w1)
query = db.query(Writer, Writer.book_count)
print(str(query)) # checking query
print()
writers: List[Writer] = query.all() # testing query
for writer, book_count in writers:
print(f"{writer.name}: {book_count}")
结果：
> Writer 1: 2
> Writer 2: 1
I'm unsure how to further add tables to the join.
SQL 从这里 db.query(Writer, Writer.book_count) 看起来很干净，没有任何连接。所以，我认为你应该不会对后续加入有任何问题。
> SELECT writer.id AS writer_id, writer.name AS writer_name, (SELECT count(book_writer_association.book_id) AS count_1
> FROM book_writer_association
> WHERE book_writer_association.writer_id = writer.id) AS book_count
> FROM writer
编辑：如果您需要加入Book table以提供额外的过滤，您可以这样做。这里我筛选了价格低于150
from typing import List
from sqlalchemy import Column, ForeignKey, Integer, String, select, func, create_engine, Table
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, object_session, sessionmaker, Session
# Declare models
@as_declarative()
class Base:
pass
book_writer_association_table = Table('book_writer_association',Base.metadata,
Column('book_id',ForeignKey('book.id'), primary_key=True),
Column('writer_id',ForeignKey('writer.id'), primary_key=True)
)
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True)
name = Column(String)
price = Column(Integer)
writers = relationship("Writer", secondary=book_writer_association_table, back_populates="books")
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship("Book", secondary=book_writer_association_table, back_populates="writers")
@hybrid_property
def book_count(self):
return (
object_session(self)
.query(book_writer_association_table)
.join(Book, Book.id == book_writer_association_table.c.book_id)
.filter(book_writer_association_table.c.writer_id == self.id)
.filter(Book.price > 150)
.count()
)
@book_count.expression
def book_count(cls):
# return select([func.count(book_writer_association_table.c.book_id)]).where(book_writer_association_table.c.writer_id == cls.id).label('book_count')
#
return (
select([func.count(book_writer_association_table.c.book_id)])
.join(Book, Book.id == book_writer_association_table.c.book_id)
.where(book_writer_association_table.c.writer_id == cls.id)
.filter(Book.price > 150)
.label('book_count')
)
# Load DB schema
engine = create_engine('sqlite:///sqlite3.db')
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=True, bind=engine)
db: Session = SessionLocal()
# Creating test instances
b1 = Book(name="Book 1", price=100)
b2 = Book(name="Book 2", price=200)
db.add(b1)
db.add(b2)
w1 = Writer(name="Writer 1")
w2 = Writer(name="Writer 2")
db.add(w1)
db.add(w2)
b1.writers.append(w1)
b1.writers.append(w2)
b2.writers.append(w1)
query = db.query(Writer, Writer.book_count)
print(str(query)) # checking query
print()
writers: List[Writer] = query.all() # testing query
for writer, book_count in writers:
print(f"{writer.name}: {book_count}")
查询：
SELECT writer.id AS writer_id,
writer.name AS writer_name,
(SELECT count(book_writer_association.book_id) AS count_1
FROM book_writer_association
JOIN book ON book.id = book_writer_association.book_id
WHERE book_writer_association.writer_id = writer.id
AND book.price > ?) AS book_count
FROM writer