SQLAlchemy 多表和外键连接

Question

有四张桌子； users、company、company_branch 和 users_branch。用户是属于公司的人。公司有分支机构，用户可以在任何给定时间属于单个分支机构。但是，users_branch 表的存在是为了跟踪从一个分支更改到另一个分支的历史记录。例如。要获取 id 为 1 的用户的当前分支，可以运行 SELECT company_id, company_branch_id FROM users_branch WHERE user_id = 1 ORDER BY created_at DESC LIMIT 1。

我面临的挑战是，我无法找出正确的非 SQLAlchemy ORM 语法，但也无法找出 SQL 原始来获取特定公司在给定时间的用户列表，并在为每个条目返回 users_id, users_email_address, company_id, company_name, compancy_branch_id and company_branch_name 时这样做.到目前为止，我尝试过的查询要么不返回任何内容，要么在 users_branch 中返回重复值，因为我只想要每个用户的最新分支

Here is the link 到 sqlfiddle 示例 postgresql 数据库。在 SQAlchemy 中，模型为 Users, Company, CompanyBranch, UsersBranch，如下所示：

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    email_address = Column(String(70), nullable=False, unique=True)

class Company(Base):
    __tablename__ = 'company'
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    created_by = Column(ForeignKey('users.id'), nullable=False)
    company_name = Column(String(100), nullable=False, unique=True)

class CompanyBranch(Base):
    __tablename__ = 'company_branch'
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    created_by = Column(ForeignKey('users.id'), nullable=False)
    company_id = Column(ForeignKey('company.id'), nullable=False)
    branch_name = Column(String(100), nullable=False, unique=True)

class UsersBranch(Base):
    __tablename__ = 'users_branch'
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    created_by = Column(ForeignKey('users.id'), nullable=False)
    user_id = Column(ForeignKey('users.id'), nullable=False)
    company_id = Column(ForeignKey('company.id'), nullable=False)
    company_branch_id = Column(ForeignKey('company_branch.id'), nullable=False)

Answer 1

首先，让我先说您的架构有点非规范化。 users_branch.company_id 不是必需的，因为users_branch.company_branch_id 也可以为您提供company_id。这样做可能有充分的理由，但它可能会在这里增加一些混乱。

由于users_branch 表，这很棘手。它本质上需要按user_id 分组并选择最大created_at 的行。

SELECT DISTINCT ON (users_branch.user_id)
  *
FROM
  users
  JOIN users_branch ON users.id = users_branch.user_id
  JOIN company_branch ON users_branch.company_branch_id = company_branch.id
  JOIN company ON company_branch.company_id = company.id
WHERE users_branch.created_at < [some date]
ORDER BY users_branch.user_id, users_branch.created_at DESC;

不过，这并不能很好地映射到 SQLAlchemy ORM。

Answer 2

我想我已经找到了我需要的东西。以下原始 SQL 代码似乎给了我正确的答案，即只返回用户所在的当前分支。花了我一段时间，但我也想出了 SQlAlchemy 的等价物。我会暂时把它留在这里作为答案，看看是否有其他人可以进一步调整它。

原始 SQL

SELECT DISTINCT ON (users_branch.user_id) users.email_address, company.id as company_id, company.company_name, company_branch.id AS company_branch_id, company_branch.branch_name
FROM
  users
  JOIN users_branch ON users.id = users_branch.user_id
  JOIN company_branch ON users_branch.company_branch_id = company_branch.id
  JOIN company ON company_branch.company_id = company.id
WHERE users_branch.created_at in (SELECT max(users_branch.created_at) FROM users_branch GROUP BY users_branch.user_id) AND 
users_branch.company_id = 1 AND
users_branch.company_branch_id = 3

SQL 炼金术

query = session.query(Users.id.label('user_id'), Users.email_address, Company.id.label('company_id'), Company.company_name,
CompanyBranch.id.label("company_branch_id"), CompanyBranch.branch_name).distinct(UsersBranch.user_id). \
join(UsersBranch, and_(Users.id == UsersBranch.user_id)). \
join(CompanyBranch, and_(UsersBranch.company_branch_id == CompanyBranch.id)).\
join(Company, and_(CompanyBranch.company_id == Company.id)).\
filter(UsersBranch.created_at.in_(session.query(func.max(UsersBranch.created_at)).group_by(UsersBranch.user_id))).\
filter(UsersBranch.company_id == 1).\
filter(UsersBranch.company_branch_id == 3)