python - SQLAlchemy 过滤不工作

Question

我一直在尝试对查询使用过滤器，但由于某种原因，过滤似乎不起作用。例如，如果我运行命令:

Curriculum_Version.query.filter(Course.course_code == 'PIP-001').all()

我得到的结果与我运行的结果相同:

Curriculum_Version.query.filter(Course.course_code == 'FEWD-001').all()

(均返回):

[#1 Version Number: 1, Date Implemented: 2013-07-23 00:00:00, #2 Version Number: 2, Date Implemented: 2013-07-24 00:00:00]

如果我运行:

Curriculum_Version.query.get(1).course

我得到:

from main import app, db
from flask import Flask, request, g, redirect, url_for
from flaskext.auth import Auth, AuthUser, login_required, get_current_user_data
from flaskext.auth.models.sa import get_user_class
import datetime
from flask.ext.sqlalchemy import SQLAlchemy
import pdb


class User(db.Model, AuthUser):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    tf_login = db.Column(db.String(255), unique=True, nullable=False)  # can assume is an email
    password = db.Column(db.String(120), nullable=False)
    salt = db.Column(db.String(80))
    role = db.Column(db.String(80))  # for later when have different permission types
    zoho_contactid = db.Column(db.String(20), unique=True, nullable=False)
    created_asof = db.Column(db.DateTime, default=datetime.datetime.utcnow)
    firstname = db.Column(db.String(80))
    lastname = db.Column(db.String(80))

    def __init__(self, zoho_contactid, firstname, lastname, tf_login, password, role, *args, **kwargs):
        super(User, self).__init__(tf_login=tf_login, password=password, *args, **kwargs)
        if (password is not None) and (not self.id):
            self.created_asof = datetime.datetime.utcnow()
            # Initialize and encrypt password before first save.
            self.set_and_encrypt_password(password)
        self.zoho_contactid = zoho_contactid  # TODO
        self.firstname = firstname
        self.lastname = lastname
        self.tf_login = tf_login  # TODO -- change to tf_login
        self.role = role

    def __repr__(self):
        return '#%d tf_login: %s, First Name: %s Last Name: %s created_asof %s' % (self.id, self.tf_login, self.firstname, self.lastname, self.created_asof)

    def __getstate__(self):
        return {
            'id': self.id,
            'tf_login': self.tf_login,
            'firstname': self.firstname,
            'lastname': self.lastname,
            'role': self.role,
            'created_asof': self.created_asof,
        }

    def __eq__(self, o):
        return o.id == self.id

    @classmethod
    def load_current_user(cls, apply_timeout=True):
        data = get_current_user_data(apply_timeout)
        if not data:
            return None
        return cls.query.filter(cls.email == data['email']).one()


class Enrollment(db.Model, AuthUser):
    __tablename__ = 'enrollments'
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
    user = db.relationship('User', backref='enrollments')
    curriculum_version_id = db.Column(db.Integer, db.ForeignKey('curriculum_versions.id'))
    curriculumversion = db.relationship('Curriculum_Version', backref='enrollments')
    cohort_id = db.Column(db.Integer, db.ForeignKey('cohorts.id'))
    cohort = db.relationship('Cohort', backref='enrollments')

    def __repr__(self):
        return '#%d User ID: %s Version ID: %s, Cohort ID: %s' % (self.id, self.user_id, self.curriculum_version_id, self.cohort_id)


class Cohort(db.Model, AuthUser):
    __tablename__ = 'cohorts'
    id = db.Column(db.Integer, primary_key=True)
    start_date = db.Column(db.DateTime)
    course_id = db.Column(db.Integer, db.ForeignKey('courses.id'))
    course = db.relationship('Course', backref='cohorts')

    def __repr__(self):
        return '#%d Start Date: %s, Course: %s' % (self.id, self.start_date, self.course.course_code)


class Curriculum_Version(db.Model, AuthUser):
    __tablename__ = 'curriculum_versions'
    id = db.Column(db.Integer, primary_key=True)
    version_number = db.Column(db.String(6))
    date_implemented = db.Column(db.DateTime)
    course_id = db.Column(db.Integer, db.ForeignKey('courses.id'))
    course = db.relationship('Course', backref='curriculum_versions')

    def __repr__(self):
        return '#%d Version Number: %s, Date Implemented: %s' % (self.id, self.version_number, self.date_implemented)


class Course(db.Model, AuthUser):
    __tablename__ = 'courses'
    id = db.Column(db.Integer, primary_key=True)
    course_code = db.Column(db.String(20))
    course_name = db.Column(db.String(50))

    def __repr__(self):
        return '#%d Course Code: %s, Course Name: %s' % (self.id, self.course_code, self.course_name)

    def __eq__(self, o):
        return o.id == self.id

我如何在数据库中为 Curriculum_Versions 创建条目:

def update_courses():
    course_code = request.form['course_code']
    start_date = request.form['start_date']
    course_date = datetime.strptime(start_date, '%m/%d/%Y')
    curr_version = Curriculum_Version.query.filter(Course.course_code == course_code) \
        .order_by(desc('version_number')).first()

    if curr_version is None:
        next_version = 1
    else:
        next_version = int(curr_version.version_number)+1

    existing = Curriculum_Version.query.filter(Course.course_code == course_code) \
        .filter(Curriculum_Version.date_implemented == course_date)

    if len(existing.all()) > 0:
        return "You tried to make version %d of the curriculum, but version \
        %s of the curriculum already exists for %s for class %s." \
        %(next_version, existing.first().version_number, start_date, course_code)

    course_object = Course.query.filter(Course.course_code == course_code).first()

    if course_object is None:
        return "The course %s does not yet exist!" % (course_code)

    new_version = Curriculum_Version(version_number=next_version, date_implemented=course_date, course=course_object)
    db.session.add(new_version)
    db.session.commit()
    return 'Created version %d for course %s starting on %s.' \
            %(next_version, course_code, start_date)

Answer 1

我认为您需要在过滤和仅使用一个查询之前加入:

# Get all the versions of a single course.
versions = Curriculum_Version.query.join(Curriculum_Version.course).filter(
    Course.course_code == "PIP-001").all()

否则 sqlalchemy 将不知道在过滤之前使用关系。

如果您只是指定一个过滤器，则 sqlalchemy 不知道执行连接，您最终会得到类似于以下的 sql:

SELECT curriculum_versions.* FROM curriculum_versions, courses WHERE 
    courses.course_code = "PIP-001"

这没有多大意义，但却是有效的 SQL。当您使用联接时，它会像这样针对正确的表利用过滤器:

SELECT curriculum_versions.* FROM curriculum_versions JOIN courses ON
    curriculum_versions.course_id = courses.id WHERE courses.course_code = "PIP-001"

请注意，sqlalchemy 知道使用条件 curriculum_versions.course_id = courses.id因为您将 Curriculum_Version.course 传递给 query.join() 并在您的 Curriculum_Version 类中指定该关系作为类(class)属性，它会自动知道使用 curriculum_versions 和 courses 表之间唯一可用的外键(您还必须在 curriculum_versions.course_id 列中指定)。

您可以在此处阅读有关联接的更多信息: http://docs.sqlalchemy.org/en/rel_0_7/orm/tutorial.html#querying-with-joins