5

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

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)
4

3 回答 3

4

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

# 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作为课程属性，并且它自动知道使用curriculum_versions和courses表之间唯一可用的外键（您还必须在curriculum_versions.course_id列上指定） ）。

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

于 2013-07-24T18:32:09.833 回答
2

可以如下查询：

course_id = Course.query.filter(course_code="PIP-001").first().id
curriculum = Curriculum_Version.query.filter(course_id=course_id).all()
于 2013-07-23T21:00:32.503 回答
-1

看看我放在一起的这个演示应用程序。这是一个单页 Flask 和 SQLAlchemy 应用程序，我认为它描述了您应该需要的一切：

https://github.com/Thinkful/sqlalchemy-demo

于 2013-07-25T02:48:32.797 回答