python - sqlalchemy:联合查询多个表中有条件的几列

Question

我正在尝试使 MySQLdb 应用程序的某些部分适应声明性基础中的 sqlalchemy。我只是从 sqlalchemy 开始。

遗留表定义如下:

student: id_number*, semester*, stateid, condition, ...
choice: id_number*, semester*, choice_id, school, program, ...

我们每个都有 3 个表(student_tmp、student_year、student_summer、choice_tmp、choice_year, choice_summer), 所以每对 (_tmp, _year, _summer) 包含信息特定时刻。

select *
from `student_tmp`
    inner join `choice_tmp` using (`id_number`, `semester`)

我的问题是对我来说重要的信息是获得与以下选择等效的信息:

SELECT t.*
FROM (
        (
            SELECT st.*, ct.*
            FROM `student_tmp` AS st
                INNER JOIN `choice_tmp` as ct USING (`id_number`, `semester`)
            WHERE (ct.`choice_id` = IF(right(ct.`semester`, 1)='1', '3', '4'))
                AND (st.`condition` = 'A')
        ) UNION (
            SELECT sy.*, cy.*
            FROM `student_year` AS sy
                INNER JOIN `choice_year` as cy USING (`id_number`, `semester`)
            WHERE (cy.`choice_id` = 4)
                AND (sy.`condition` = 'A')
        ) UNION (
            SELECT ss.*, cs.*
            FROM `student_summer` AS ss
                INNER JOIN `choice_summer` as cs USING (`id_number`, `semester`)
            WHERE (cs.`choice_id` = 3)
                AND (ss.`condition` = 'A')
        )
    ) as t

* 用于缩短选择，但我实际上只查询 50 个可用列中的大约 7 个列。

此信息有多种用途...“我有新学生吗？我还有给定日期的所有学生吗？哪些学生在给定日期之后订阅了？等等...”这个选择的结果语句将被保存在另一个数据库中。

我可以用一个类似 View 的类来实现吗？该信息是只读的，因此我不需要能够修改/创建/删除。或者我是否必须为每个表声明一个类(最终有 6 个类)并且每次我需要查询时，我都必须记住过滤？

多谢指点。

编辑:我没有对数据库的修改权限(我无法创建 View )。这两个数据库可能不在同一台服务器上(因此我无法在我的第二个数据库上创建 View )。

我担心的是在根据 condition 和 choice_id 进行过滤之前避免全表扫描。

编辑 2:我设置了这样的声明类:

class BaseStudent(object):
    id_number = sqlalchemy.Column(sqlalchemy.String(7), primary_key=True)
    semester = sqlalchemy.Column(sqlalchemy.String(5), primary_key=True)
    unique_id_number = sqlalchemy.Column(sqlalchemy.String(7))
    stateid = sqlalchemy.Column(sqlalchemy.String(12))
    condition = sqlalchemy.Column(sqlalchemy.String(3))

class Student(BaseStudent, Base):
    __tablename__ = 'student'

    choices = orm.relationship('Choice', backref='student')

#class StudentYear(BaseStudent, Base):...
#class StudentSummer(BaseStudent, Base):...

class BaseChoice(object):
    id_number = sqlalchemy.Column(sqlalchemy.String(7), primary_key=True)
    semester = sqlalchemy.Column(sqlalchemy.String(5), primary_key=True)
    choice_id = sqlalchemy.Column(sqlalchemy.String(1))
    school = sqlalchemy.Column(sqlalchemy.String(2))
    program = sqlalchemy.Column(sqlalchemy.String(5))


class Choice(BaseChoice, Base):
    __tablename__ = 'choice'

    __table_args__ = (
            sqlalchemy.ForeignKeyConstraint(['id_number', 'semester',],
                [Student.id_number, Student.semester,]),
            )

#class ChoiceYear(BaseChoice, Base): ...
#class ChoiceSummer(BaseChoice, Base): ...

现在，为一组表提供正确 SQL 的查询是:

q = session.query(StudentYear, ChoiceYear) \
            .select_from(StudentYear) \
            .join(ChoiceYear) \
            .filter(StudentYear.condition=='A') \
            .filter(ChoiceYear.choice_id=='4')

但是它抛出一个异常...

"Could not locate column in row for column '%s'" % key)
sqlalchemy.exc.NoSuchColumnError: "Could not locate column in row for column '*'"

我如何使用该查询为自己创建一个我可以使用的类？

Answer 1

如果您可以在数据库上创建此 View ，那么您只需将 View 映射为表即可。参见 Reflecting Views .

# DB VIEW
CREATE VIEW my_view AS -- @todo: your select statements here

# SA
my_view = Table('my_view', metadata, autoload=True)
# define view object
class ViewObject(object):
    def __repr__(self):
        return "ViewObject %s" % str((self.id_number, self.semester,))
# map the view to the object
view_mapper = mapper(ViewObject, my_view)

# query the view
q = session.query(ViewObject)
for _ in q:
    print _

如果您不能在数据库级别创建一个VIEW，您可以创建一个可选对象并将ViewObject 映射到它。下面的代码应该给你的想法:

student_tmp = Table('student_tmp', metadata, autoload=True)
choice_tmp = Table('choice_tmp', metadata, autoload=True)
# your SELECT part with the columns you need
qry = select([student_tmp.c.id_number, student_tmp.c.semester, student_tmp.stateid, choice_tmp.school])
# your INNER JOIN condition
qry = qry.where(student_tmp.c.id_number == choice_tmp.c.id_number).where(student_tmp.c.semester == choice_tmp.c.semester)
# other WHERE clauses
qry = qry.where(student_tmp.c.condition == 'A')

您可以像这样创建 3 个查询，然后将它们与 union_all 合并并在映射器中使用生成的查询:

view_mapper = mapper(ViewObject, my_combined_qry)

在这两种情况下，您都必须确保在 View 上正确定义了主键，并且您可能需要覆盖自动加载的 View ，并明确指定主键(参见上面的链接) .否则，您将收到错误消息，或者可能无法从查询中获得正确的结果。

---

对 EDIT-2 的回答:

qry = (session.query(StudentYear, ChoiceYear).
        select_from(StudentYear).
        join(ChoiceYear).
        filter(StudentYear.condition == 'A').
        filter(ChoiceYear.choice_id == '4')
        )

结果将是元组对:(Student, Choice)。
但是如果你想为查询创建一个新的映射类，那么你可以像上面的示例一样创建一个可选的:

student_tmp = StudentTmp.__table__
choice_tmp = ChoiceTmp.__table__
.... (see sample code above)