通过关联建立多对多关系

Question

我有以下型号：

群组：

class Group(db.Model):
    __tablename__ = 'Groups'
    __table_args__ = {'extend_existing': True}
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(120), index=True, unique=True)
    description = db.Column(db.Text, nullable=False)
    created_on = db.Column(db.DateTime, default=datetime.now)
    updated_on = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
    users = db.relationship(
        "User",
        secondary=users_groups_assocation_table,
        back_populates="groups")
    analysis = db.relationship(
        "Analysis",
        secondary=analysis_groups_assocation_table,
        back_populates="groups"

用户：

class User(db.Model):
    __tablename__ = 'Users'
    __table_args__ = {'extend_existing': True}
    id = db.Column(db.Integer, primary_key=True)
    alias = db.Column(db.String(120), index=True, unique=True)
    name = db.Column(db.String(120), index=True, unique=False)
    lastname = db.Column(db.String(120), index=True, nullable=False)
    email = db.Column(db.String(120), index=True, nullable=False)
    password = db.Column(db.String(120), index=False, nullable=True)
    created_on = db.Column(db.DateTime, default=datetime.now)
    updated_on = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
    admin = db.Column(db.Boolean, default=False)
    local_auth = db.Column(db.Boolean, default=False)
    override_tableau = db.Column(db.Boolean, default=False)
    groups = db.relationship(
        "Group",
        secondary=users_groups_assocation_table,
        back_populates="users")

分析：

class Analysis(db.Model):
  __tablename__ = 'Analysis'
  __table_args__ = {'extend_existing': True}
  id = db.Column(db.Integer, primary_key=True)
  name = db.Column(db.String(120), index=True, unique=True)
  description = db.Column(db.Text, nullable=False)
  embed_analysis = db.Column(db.Text, nullable=True)
  service_account = db.Column(db.Text, nullable=True)
  img = db.Column(db.LargeBinary(length=(2**32)-1), nullable=True)
  img_mimetype = db.Column(db.Text, nullable=True)
  img_name = db.Column(db.Text, nullable=True)
  category_id = db.Column(db.Integer, db.ForeignKey('Categories.id'), nullable=True)
  created_on = db.Column(db.DateTime, default=datetime.now)
  updated_on = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
  category = db.relationship("Category", back_populates="analysis")
  draft = db.Column(db.Boolean, default=False)
  groups = db.relationship(
        "Group",
        secondary=analysis_groups_assocation_table,
        back_populates="analysis")

关联表：

users_groups_assocation_table = db.Table('users_groups',
    db.Column('user_id', db.Integer, db.ForeignKey('Users.id')),
    db.Column('group_id', db.Integer, db.ForeignKey('Groups.id'))
)

analysis_groups_assocation_table = db.Table('analysis_groups',
    db.Column('analysis_id', db.Integer, db.ForeignKey('Analysis.id')),
    db.Column('group_id', db.Integer, db.ForeignKey('Groups.id'))
)

所以在用户<->分析之间存在隐含的多对多关系。如何通过 groups 创建这样的关联？来自 ruby on rails 在这种情况下有一个 :through 关键字，flask sqlalchemy 是否有类似的东西？

我想要 User.query.first().analysis

Answer 1

sqlalchemy 的关系可以做很多事情，但我发现复杂的关系很难获得 right/predictable 并且一次性查询更容易维护（至少对于复杂的关系）。

我试图解决您询问的关系，如 User.analyses 但在末尾包含了一个一次性查询。

我在此处的文档中使用了跨多个表的简化版本：https://docs.sqlalchemy.org/en/14/orm/join_conditions.html#composite-secondary-join

请注意，连接混合使用了 Table 对象，这些对象需要引用 .c. 的列并映射类，例如直接引用列的 Analysis。

您可能还需要 relationship 的 order_by 参数，否则这种关系可能毫无意义。

from datetime import datetime, date
from sqlalchemy import (
    create_engine,
    Text,
    Integer,
    String,
    ForeignKey,
    UniqueConstraint,
    update,
    DateTime,
   Date,
    Boolean,
    LargeBinary,
)
from sqlalchemy.schema import (
    Table,
    Column,
    MetaData,
)
from sqlalchemy.sql import select
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError


Base = declarative_base()


engine = create_engine("sqlite://", echo=False)

users_groups_table = Table('users_groups', Base.metadata,
    Column('user_id', Integer, ForeignKey('Users.id')),
    Column('group_id', Integer, ForeignKey('Groups.id'))
)

analysis_groups_table = Table('analysis_groups', Base.metadata,
    Column('analysis_id', Integer, ForeignKey('Analysis.id')),
    Column('group_id', Integer, ForeignKey('Groups.id'))
)

class Group(Base):
    __tablename__ = 'Groups'
    __table_args__ = {'extend_existing': True}
    id = Column(Integer, primary_key=True)
    name = Column(String(120), index=True, unique=True)
    users = relationship(
        "User",
        secondary=users_groups_table,
        back_populates="groups")
    analysis = relationship(
        "Analysis",
        secondary=analysis_groups_table,
        back_populates="groups")

class User(Base):
    __tablename__ = 'Users'
    __table_args__ = {'extend_existing': True}
    id = Column(Integer, primary_key=True)
    name = Column(String(120), index=True, unique=False)
    groups = relationship(
        "Group",
        secondary=users_groups_table,
        back_populates="users")
    analyses = relationship("Analysis",
                            # The middle
                            secondary="join(users_groups, analysis_groups, users_groups.c.group_id == analysis_groups.c.group_id)",
                            # Join from left to the middle
                            primaryjoin="User.id == users_groups.c.user_id",
                            # Join from right to the middle
                            secondaryjoin="Analysis.id == analysis_groups.c.analysis_id",
                uselist=True,
                viewonly=True
                )

class Analysis(Base):
  __tablename__ = 'Analysis'
  __table_args__ = {'extend_existing': True}
  id = Column(Integer, primary_key=True)
  name = Column(String(120), index=True, unique=True)
  groups = relationship(
        "Group",
        secondary=analysis_groups_table,
        back_populates="analysis")



Base.metadata.create_all(engine)

with Session(engine) as session:
    users = {}
    for name in ('a', 'b'):
        users[name] = User(name=name)
        session.add(users[name])
    groups = {}
    for name in ('a', 'b'):
        groups[name] = Group(name=name)
        session.add(groups[name])
    analyses = {}
    for name in ('x', 'y', 'z'):
        analyses[name] = Analysis(name=name)
        session.add(analyses[name])

    groups['a'].users.append(users['a'])
    groups['b'].users.append(users['b'])
    analyses['x'].groups.append(groups['a'])
    analyses['y'].groups.append(groups['b'])
    analyses['z'].groups.append(groups['b'])

    session.commit()

    print (users['a'].analyses[0].name)

    # One-off ad-hoc query.
    q = session.query(Analysis).join(Analysis.groups).join(Group.users).filter(User.id == users['a'].id)
    print (q)
    res = q.first()
    print (res.name)

通过关联建立多对多关系

Many to many relationship through association

flask

sqlalchemy

flask-sqlalchemy