python - 在sqlalchemy中连接两个没有关系的表

Question

这是我的问题。我有三张 table 。

一个名为 Project 的项目，只有一个名为 id 的列(该列在整个系统中必须是唯一的)。
一个称为 ServiceAwarenessProject，它与 Project.id 具有一对一的关系。
名为 CorporateVPNProject 的项目与 Project.id 具有一对一的关系

我正在使用 sqlalchemy ORM，因此代码如下所示:

class Project(SqlAlchemyBase):
    __tablename__ = 'project'

    id = Column(Integer, primary_key=True, autoincrement=True)


class ServiceAwarenessProject(SqlAlchemyBase):
    __tablename__ = 'sa_project'

    id = Column(Integer, primary_key=True)
    project_id = Column(Integer, ForeignKey(Project.id))
    mop_url = Column(String, nullable=False)
    expiration_date = Column(Datetime, index=True)


class CorporateVPNProject(SqlAlchemyBase):
    __tablename__ = 'wvpn_project'

    id = Column(Integer, primary_key=True)
    project_id = Column(Integer, ForeignKey(Project.id))
    mop_url = Column(String, nullable=False)

我这样设计了我的表，所以我可以保证我在整个系统中拥有唯一的project_ids。我的问题是我不知道如何将这些表连接在一起以根据project_id查找项目。为了暂时解决这个问题，我使用名为 get_project_by_id 的函数查询两个表。有更聪明的方法来解决这个问题吗？

class ProjectService:
    @staticmethod
    def create_project_id():
        session = DbSessionFactory.create_session()
        result = session.query(Project.id).order_by(desc(Project.id)).first()

        if result:
            result = result[0]
            if str(result)[:8] == datetime.datetime.now().strftime('%Y%m%d'):
                project_id = str(result)[:8] + '{:03d}'.format(int(str(result)[8:]) + 1)
                new_project = Project(id=project_id)
                session.add(new_project)
                session.commit()
                return project_id

        project_id = datetime.datetime.now().strftime('%Y%m%d') + '001'
        new_project = Project(id=project_id)
        session.add(new_project)
        session.commit()
        return project_id

    @staticmethod
    def get_project_by_id(project_id):
        session = DbSessionFactory.create_session()
        result = session.query(ServiceAwarenessProject) \
            .filter(ServiceAwarenessProject.project_id == project_id) \
            .first()

        if result:
            return result

        result = session.query(CorporateVPNProject) \
            .filter(CorporateVPNProject.project_id == project_id) \
            .first()
        if result:
            return result

    def create_serviceawareness_project(self):
        session = DbSessionFactory.create_session()
        project_id = self.create_project_id()
        new_project = ServiceAwarenessProject(project_id=project_id, mop_url='http://www.thepacketwizards.com/1')
        session.add(new_project)
        session.commit()
        return new_project

    def create_corporatevpn_project(self):
        session = DbSessionFactory.create_session()
        project_id = self.create_project_id()
        new_project = CorporateVPNProject(project_id=project_id, mop_url='http://www.thepacketwizards.com/wvpn')
        session.add(new_project)
        session.commit()
        return new_project

谢谢!

Answer 1

根据@Ilja Everilä的建议，我像这样设计了表格，仅使用joined table inheritance .

class Project(SqlAlchemyBase):
    __tablename__ = 'project'

    id = Column(Integer, primary_key=True)
    created_on = Column(DateTime, default=datetime.datetime.now)
    updated_on = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
    project_url = Column(String(60))
    mop_url = Column(String(60))
    input_url = Column(String(60))
    type = Column(String(60))
    __mapper_args__ = {
        'polymorphic_identity': 'project',
        'polymorphic_on': type
    }


class ServiceAwarenessProject(Project):
    __tablename__ = 'sa_project'

    id = Column(Integer, ForeignKey('project.id'), primary_key=True)
    expiration_date = Column(DateTime)
    __mapper_args__ = {
        'polymorphic_identity': 'ServiceAwareness',
    }


class CorporateVPNProject(Project):
    __tablename__ = 'wvpn_project'

    id = Column(Integer, ForeignKey('project.id'), primary_key=True)
    client_name = Column(String(60))
    __mapper_args__ = {
        'polymorphic_identity': 'CorporateVPN',
    }

现在，要查询数据库，我必须使用 with_polymorphic，这样我就可以获得每行的不同表实例。

class ProjectService:
    @staticmethod
    def create_project_id():
        session = DbSessionFactory.create_session()
        result = session.query(Project.id).order_by(desc(Project.id)).first()
        print(result)
        if result:
            result = result[0]
            if str(result)[:8] == datetime.datetime.now().strftime('%Y%m%d'):
                project_id = str(result)[:8] + '{:03d}'.format(int(str(result)[8:]) + 1)
                return project_id

        project_id = datetime.datetime.now().strftime('%Y%m%d') + '001'
        return project_id

    def create_serviceawareness_project(self):
        session = DbSessionFactory.create_session()
        project_id = self.create_project_id()
        new_project = ServiceAwarenessProject(id=project_id,
                                              project_url='http://project',
                                              expiration_date=datetime.datetime.now() + datetime.timedelta(days=365),
                                              mop_url='http://mop',
                                              input_url='http://url',
                                              type='ServiceAwareness')

        session.add(new_project)
        session.commit()
        session.add(new_project)

        return new_project

    def create_corporatevpn_project(self):
        session = DbSessionFactory.create_session()
        project_id = self.create_project_id()
        new_project = CorporateVPNProject(id=project_id,
                                          project_url='http://project',
                                          client_name='TIM',
                                          mop_url='http://mop',
                                          input_url='http://url',
                                          type='CorporateVPN')

        session.add(new_project)
        session.commit()
        session.add(new_project)

        return new_project

    @staticmethod
    def get_project_by_id(project_id):
        session = DbSessionFactory.create_session()
        query = session.query(with_polymorphic(Project, [ServiceAwarenessProject, CorporateVPNProject])).filter(or_(
            ServiceAwarenessProject.id == project_id,
            CorporateVPNProject.id == project_id
        )).first()
        return query