最新消息:20210917 已从crifan.com换到crifan.org

【已解决】Flask中SQLAlchemy去实现查询后的分页结果的排序

Flask crifan 3666浏览 0评论

之前已经实现了查询和分页

            taskPagination = Task.query.filter(Task.initiatorId==userId).paginate(
                page=curPageNum,
                per_page=numPerPage,
                error_out=False)

但是返回的结果,是没有排序的,导致:

最旧的Task排在最前

现在希望实现query后,去排序

或者query后的pagination后,去排序

根据时间,最新的放在最开始

sqlalchemy query sort

sqlalchemy pagination sort

Ordering your sqlalchemy many-to-many relationship – julo

order_by

poor performance of pagination under large dataset · Issue #272 ..

order_by( Info.create_time.desc()).paginate(page, per_page, True)

SQLAlchey order_by

Python SQLAlchemy基本操作和常用技巧(包含大量实例,非常好)_python_脚本之家

SQLAlchemy 使用经验 – tastelife的专栏 – 博客频道 – CSDN.NET

print query2.order_by(User.name).all()
print query2.order_by(‘name’).all()
print query2.order_by(User.name.desc()).all()
print query2.order_by(‘name desc’).all()
print session.query(User.id).order_by(User.name.desc(), User.id).all()

不过此处想要去研究:

【已解决】SQLAlchemy中order_by默认是否是升序以及写法是否是asc

然后用代码:

        curPageTaskList = None
        filterTask = None
        if curRole == UserRole.Initiator:
            filterTask = Task.query.filter(Task.initiatorId==userId)
        elif curRole == UserRole.Errandor:
            filterTask = Task.query.filter(Task.errandorId == userId)
        # gLog.debug(“filterTask=%s”, filterTask)
        descOrderByTask = filterTask.order_by(Task.createdAt.desc())
        # gLog.debug(“descOrderByTask=%s”, descOrderByTask)
        taskPagination = descOrderByTask.paginate(page=curPageNum, per_page=numPerPage, error_out=False)
        gLog.debug(“type(taskPagination)=%s”
                   “,taskPagination=%s”
                   “,has_next=%s”
                   “,has_prev=%s”
                   # “,items=%s”
                   # “,next()=%s”
                   # “,next_num=%s”
                   “,page=%s”
                   “,pages=%s”
                   “,per_page=%s”
                   # “,prev_num=%s”
                   # “,query=%s”
                   “,total=%s”,
                   type(taskPagination),
                   taskPagination,
                   taskPagination.has_next,
                   taskPagination.has_prev,
                   # taskPagination.items,
                   # taskPagination.next(error_out=False),
                   # taskPagination.next_num,
                   taskPagination.page,
                   taskPagination.pages,
                   taskPagination.per_page,
                   # taskPagination.prev_num,
                   # taskPagination.query,
                   taskPagination.total
                   )

对应的log是:

DEBUG in User [/root/RunningFast/staging/runningfast/resources/User.py:542]:
filterTask=SELECT tasks.id AS tasks_id, tasks.`statusType` AS `tasks_statusType`, tasks.`itemType` AS `tasks_itemType`, tasks.`promotionCode` AS `tasks_promotionCode`, tasks.`hasEnded` AS `tasks_hasEnded`, tasks.`endedTime` AS `tasks_endedTime`, tasks.`createdAt` AS `tasks_createdAt`, tasks.`updatedAt` AS `tasks_updatedAt`, tasks.`initiatorId` AS `tasks_initiatorId`, tasks.`errandorId` AS `tasks_errandorId`, tasks.`cancelByUserType` AS `tasks_cancelByUserType`, tasks.`initiatorTipType` AS `tasks_initiatorTipType`, tasks.`initiatorStartLocationId` AS `tasks_initiatorStartLocationId`, tasks.`initiatorEndLocationId` AS `tasks_initiatorEndLocationId`, tasks.`initiatorDescription` AS `tasks_initiatorDescription`, tasks.`initiatorCancelReasonType` AS `tasks_initiatorCancelReasonType`, tasks.`errandorStartLocationId` AS `tasks_errandorStartLocationId`, tasks.`errandorEndLocationId` AS `tasks_errandorEndLocationId`, tasks.`errandorDescription` AS `tasks_errandorDescription`, tasks.`errandorCancelReasonType` AS `tasks_errandorCancelReasonType`, tasks.`errandorStartTime` AS `tasks_errandorStartTime`, tasks.`errandorRatingType` AS `tasks_errandorRatingType`
FROM tasks
WHERE tasks.`initiatorId` = %s

<div–<——————————————————————————

<div–<——————————————————————————

DEBUG in User [/root/RunningFast/staging/runningfast/resources/User.py:544]:
descOrderByTask=SELECT tasks.id AS tasks_id, tasks.`statusType` AS `tasks_statusType`, tasks.`itemType` AS `tasks_itemType`, tasks.`promotionCode` AS `tasks_promotionCode`, tasks.`hasEnded` AS `tasks_hasEnded`, tasks.`endedTime` AS `tasks_endedTime`, tasks.`createdAt` AS `tasks_createdAt`, tasks.`updatedAt` AS `tasks_updatedAt`, tasks.`initiatorId` AS `tasks_initiatorId`, tasks.`errandorId` AS `tasks_errandorId`, tasks.`cancelByUserType` AS `tasks_cancelByUserType`, tasks.`initiatorTipType` AS `tasks_initiatorTipType`, tasks.`initiatorStartLocationId` AS `tasks_initiatorStartLocationId`, tasks.`initiatorEndLocationId` AS `tasks_initiatorEndLocationId`, tasks.`initiatorDescription` AS `tasks_initiatorDescription`, tasks.`initiatorCancelReasonType` AS `tasks_initiatorCancelReasonType`, tasks.`errandorStartLocationId` AS `tasks_errandorStartLocationId`, tasks.`errandorEndLocationId` AS `tasks_errandorEndLocationId`, tasks.`errandorDescription` AS `tasks_errandorDescription`, tasks.`errandorCancelReasonType` AS `tasks_errandorCancelReasonType`, tasks.`errandorStartTime` AS `tasks_errandorStartTime`, tasks.`errandorRatingType` AS `tasks_errandorRatingType`
FROM tasks
WHERE tasks.`initiatorId` = %s ORDER BY tasks.`createdAt` DESC

<div–<——————————————————————————

<div–<——————————————————————————

DEBUG in User [/root/RunningFast/staging/runningfast/resources/User.py:572]:
type(taskPagination)=<class ‘flask_sqlalchemy.Pagination’>,taskPagination=<flask_sqlalchemy.Pagination object at 0x7fbf70424cd0>,has_next=True,has_prev=False,page=1,pages=2,per_page=10,query=SELECT tasks.id AS tasks_id, tasks.`statusType` AS `tasks_statusType`, tasks.`itemType` AS `tasks_itemType`, tasks.`promotionCode` AS `tasks_promotionCode`, tasks.`hasEnded` AS `tasks_hasEnded`, tasks.`endedTime` AS `tasks_endedTime`, tasks.`createdAt` AS `tasks_createdAt`, tasks.`updatedAt` AS `tasks_updatedAt`, tasks.`initiatorId` AS `tasks_initiatorId`, tasks.`errandorId` AS `tasks_errandorId`, tasks.`cancelByUserType` AS `tasks_cancelByUserType`, tasks.`initiatorTipType` AS `tasks_initiatorTipType`, tasks.`initiatorStartLocationId` AS `tasks_initiatorStartLocationId`, tasks.`initiatorEndLocationId` AS `tasks_initiatorEndLocationId`, tasks.`initiatorDescription` AS `tasks_initiatorDescription`, tasks.`initiatorCancelReasonType` AS `tasks_initiatorCancelReasonType`, tasks.`errandorStartLocationId` AS `tasks_errandorStartLocationId`, tasks.`errandorEndLocationId` AS `tasks_errandorEndLocationId`, tasks.`errandorDescription` AS `tasks_errandorDescription`, tasks.`errandorCancelReasonType` AS `tasks_errandorCancelReasonType`, tasks.`errandorStartTime` AS `tasks_errandorStartTime`, tasks.`errandorRatingType` AS `tasks_errandorRatingType`
FROM tasks
WHERE tasks.`initiatorId` = %s ORDER BY tasks.`createdAt` DESC,total=19

可见:

加了排序:

.order_by(Task.createdAt.desc())

生成的查询语句是:

ORDER BY tasks.`createdAt` DESC

【总结】

最后用:

        filterTask = None
        if curRole == UserRole.Initiator:
            filterTask = Task.query.filter(Task.initiatorId==userId)
        elif curRole == UserRole.Errandor:
            filterTask = Task.query.filter(Task.errandorId == userId)
       descOrderByTask = filterTask.order_by(Task.createdAt.desc())
        taskPagination = descOrderByTask.paginate(page=curPageNum, per_page=numPerPage, error_out=False)
        taskInfoList = []
        for curIdx, eachTask in enumerate(paginatedTaskList):
            # gLog.debug(“[%s] eachTask=%s”, curIdx, eachTask)
            gLog.debug(“[%s] eachTask.id=%s”, curIdx, eachTask.id)
            taskInfoList.append(marshal(eachTask, task_fields))

实现了排序结果的分页。

注:

对应的加了排序代码:

.order_by(Task.createdAt.desc())

内部所生成的SQL查询语句是:

ORDER BY tasks.`createdAt` DESC

转载请注明:在路上 » 【已解决】Flask中SQLAlchemy去实现查询后的分页结果的排序

发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
92 queries in 0.209 seconds, using 23.45MB memory