Python app.db 模块，or_() 实例源码

def index():
    search_word = request.args.get('search', None)
    search_form = SearchForm()
    page = request.args.get('page', 1, type=int)

    the_books = Book.query
    if not current_user.can(Permission.UPDATE_BOOK_INFORMATION):
        the_books = Book.query.filter_by(hidden=0)

    if search_word:
        search_word = search_word.strip()
        the_books = the_books.filter(db.or_(
            Book.title.ilike(u"%%%s%%" % search_word), Book.author.ilike(u"%%%s%%" % search_word), Book.isbn.ilike(
                u"%%%s%%" % search_word), Book.tags.any(Tag.name.ilike(u"%%%s%%" % search_word)), Book.subtitle.ilike(
                u"%%%s%%" % search_word))).outerjoin(Log).group_by(Book.id).order_by(db.func.count(Log.id).desc())
        search_form.search.data = search_word
    else:
        the_books = Book.query.order_by(Book.id.desc())

    pagination = the_books.paginate(page, per_page=8)
    result_books = pagination.items
    return render_template("book.html", books=result_books, pagination=pagination, search_form=search_form,
                           title=u"????")

def estimate_eligible_for_delete(self):
        """
        this is an estimation because we do not know if favourite status has
        changed since last time a post was refreshed and it is unfeasible to
        refresh every single post every time we need to know how many posts are
        eligible to delete
        """
        latest_n_posts = (Post.query.with_parent(self)
                          .order_by(db.desc(Post.created_at))
                          .limit(self.policy_keep_latest))
        query = (Post.query.with_parent(self)
                 .filter(Post.created_at <=
                         db.func.now() - self.policy_keep_younger)
                 .except_(latest_n_posts))
        if(self.policy_keep_favourites):
            query = query.filter(db.or_(Post.favourite == False, Post.is_reblog))
        if(self.policy_keep_media):
            query = query.filter(db.or_(Post.has_media == False, Post.is_reblog))
        if(self.policy_keep_direct):
            query = query.filter(~Post.direct)
        return query.count()

def perform_some_search(queryset, user_input):
    return queryset.filter(
        db.or_(
            models.Processed.title.like('%'+user_input+'%'),
            models.Processed.user.like('%'+user_input+'%'),
            models.Processed.platform.like('%'+user_input+'%')
            )
        )

def charts():
    #all10 = db.session.query(models.Processed.count as count, models.Processed).group_by(models.Processed.title).order_by(db.func.count(models.Processed.title).desc()).limit(10)
    all10 = db.session.query(db.func.count(models.Processed.title), models.Processed).group_by(models.Processed.title).having(db.func.count(models.Processed.title) > 0).order_by(db.func.count(models.Processed.title).desc(), models.Processed.time.desc()).limit(10)
    movie10 = db.session.query(db.func.count(models.Processed.title), models.Processed).filter(db.or_(models.Processed.orig_title_ep == None, models.Processed.orig_title_ep == "", models.Processed.orig_title_ep == "n/a")).group_by(models.Processed.title).having(db.func.count(models.Processed.title) > 0).order_by(db.func.count(models.Processed.title).desc(), models.Processed.time.desc()).limit(10)
    show_top10 = db.session.query(db.func.count(models.Processed.orig_title), models.Processed).group_by(models.Processed.orig_title).having(db.func.count(models.Processed.orig_title) > 0).order_by(db.func.count(models.Processed.orig_title).desc(), models.Processed.time.desc()).limit(10)
    ep_top10 = db.session.query(db.func.count(models.Processed.title), models.Processed).filter(models.Processed.orig_title_ep != "n/a").group_by(models.Processed.title).having(db.func.count(models.Processed.title) > 0).order_by(db.func.count(models.Processed.title).desc(), models.Processed.time.desc()).limit(10)

    return render_template('charts.html', all_top10=all10, movie_top10=movie10, show_top10=show_top10, ep_top10=ep_top10, title=_('Charts'))

def get_unnotified():
    logger.info(u"getting unnotified entrys from database")
    result = db.session.query(models.Processed).filter(db.or_(models.Processed.notified == None, models.Processed.notified != 1)).all()
    return result

def profile():
    user = current_user
    terms = db.or_(PointLog.sender_id==user.id, PointLog.receiver_id==user.id)
    query = PointLog.query.filter(terms).order_by(PointLog.created_at.desc())
    form = ProfileForm(request.form, name=user.name, bio=user.bio)
    if request.method == 'POST' and form.validate_on_submit():
        current_user.name = form.name.data
        current_user.bio = form.bio.data
        db.session.commit()
        flash('Profile updated successfully')
    return render_template('settings/profile.html', form=form,
                            sidebar_active='profile',
                            point_logs=query.limit(20))

def sign_up():
    form = SignUpForm()
    user = User()
    if form.validate_on_submit():
        user_name = request.form.get('user_name')
        user_email = request.form.get('user_email')
        password = request.form.get('password')
        password = generate_password_hash(password)
        register_check = User.query.filter(db.or_(
            User.nickname == user_name, User.email == user_email)).first()
        if register_check:
            flash("????????")
            return redirect('/sign-up')

        if len(user_name) and len(user_email):
            user.nickname = user_name
            user.email = user_email
            user.role = ROLE_USER
            user.password = password
            try:
                db.session.add(user)
                db.session.commit()
            except:
                flash("???????????")
                return redirect('/sign-up')

            flash("????")
            return redirect('/login')

    return render_template("sign_up.html", form=form)

def login_check(cls, user_name):
        user = cls.query.filter(db.or_(
            User.nickname == user_name, User.email == user_name)).first()

        if not user:
            return None

        return user

def get_applying(cls):
        return cls.query.filter(db.or_(cls.status == 'applying', cls.renewing == True)).order_by(cls.applytime).all()

def get_users(cls):
        return cls.query.filter(db.or_(cls.status == 'pass', cls.status == 'banned')).order_by(cls.id).all()