sqlalchemy 如何在没有外键的情况下访问 Joined Model 属性？

Question

我想加入三个Model并访问其中的所有属性，但是我没有权限添加外键。

这可以通过使用原始 sql 来解决，但我想使用 sqlalchemy 模型。

模型是从现有数据库生成的：

class Marketingplanproduct(Base):
    __tablename__ = 'marketingplanproducts'

    id = Column(String(36, 'utf8_bin'), primary_key=True, server_default=text("''"))
    price = Column(Integer, nullable=False)
    marketing_plan_id = Column(ForeignKey('marketingplans.id'), index=True)
    product_id = Column(ForeignKey('products.id'), index=True)
    is_deleted = Column(Integer, nullable=False)

    marketing_plan = relationship('Marketingplan')
    product = relationship('Product')


class Marketingplan(Base):
    __tablename__ = 'marketingplans'

    id = Column(String(36, 'utf8_bin'), primary_key=True, server_default=text("''"))
    subject = Column(String(50), nullable=False, index=True)
    description = Column(String(1000), index=True)
    time_start_plan = Column(BigInteger, nullable=False, index=True)
    time_end_plan = Column(BigInteger, nullable=False, index=True)
    product_count = Column(Integer, nullable=False)
    user_id_create = Column(String(36, 'utf8_bin'), nullable=False, server_default=text("''"))
    review_status = Column(Integer, nullable=False)
    user_id_review = Column(String(36, 'utf8_bin'), nullable=False, server_default=text("''"))
    time_review = Column(BigInteger, nullable=False)
    is_deleted = Column(Integer, nullable=False)
    time_create = Column(BigInteger, nullable=False, index=True)
    time_update = Column(BigInteger, nullable=False, index=True)
    user_id_update = Column(String(36, 'utf8_bin'), nullable=False, server_default=text("''"))
    accepted_count = Column(Integer, nullable=False)
    total_execute_log_count = Column(Integer, nullable=False)
    price_change_category = Column(Integer, nullable=False)
    store_implement = Column(Integer, nullable=False)


class Marketingplanstoremap(Base):
    __tablename__ = 'marketingplanstoremaps'

    id = Column(String(36, 'utf8_bin'), primary_key=True, server_default=text("''"))
    marketing_plan_id = Column(String(36, 'utf8_bin'), nullable=False, index=True, server_default=text("''"))
    store_id = Column(String(36, 'utf8_bin'), nullable=False, index=True, server_default=text("''"))

我的代码：

def get_marketingplans(self, store_id=None, product_id=None, start_date=None, end_date=None):
    query = self.Session().query(Marketingplanproduct)\
                        .join(Marketingplan, Marketingplanproduct.marketing_plan_id==Marketingplan.id)\
                        .join(Marketingplanstoremap, Marketingplan.id==Marketingplanstoremap.marketing_plan_id)\
                        .filter(Marketingplan.is_deleted==0)\
                        .filter(Marketingplanproduct.is_deleted==0)

    if store_id:
        query.filter(Marketingplanstoremap.store_id==store_id)

    if product_id:
        query.filter(Marketingplanproduct.store_id==product_id)

    if start_date:
        s = ensure_millisecond(start_date)
        query.filter(Marketingplan.time_start_plan>=s)

    if end_date:
        e = ensure_millisecond(end_date)
        query.filter(Marketingplan.time_start_plan<e)

    marketingplans = query.all()
    df = pd.DataFrame([ (mp.store_id, mp.marketing_plan.product_id, mp.price) for mp in marketingplans], columns=['store_id', 'product_id', 'price'])
    return df

我的代码失败了，因为没有 mp.store_id 。

Answer 1

您似乎没有配置这样的 ORM 关系路径，您可以通过Marketingplanproduct 访问Marketingplanstoremap。由于您已经使用连接将两者都包含在查询中，因此您可以简单地将 Marketingplanstoremap 添加为查询中的第二个实体：

query = self.Session().query(Marketingplanproduct, Marketingplanstoremap)\
    ...

...
results = query.all()
df = pd.DataFrame([(mpsm.store_id, mpp.product_id, mpp.price)
                   for mpp, mpsm in results],
                  columns=['store_id', 'product_id', 'price'])

或者你可以直接询问你需要的属性：

query = self.Session().query(Marketingplanstoremap.store_id,
                             Marketingplanproduct.product_id,
                             Marketingplanproduct.price)\
    select_from(Marketingplanproduct)\
    ...

results = query.all()
df = pd.DataFrame(results, columns=['store_id', 'product_id', 'price'])