feat(pagination): replace with database level limit

Author: amakerlife

README.md

@@ -41,8 +41,6 @@ flask run
 
 学生、教师自动选择合适的登录方式，并保存
 
-在数据库层面使用 limit 和 offset 分页
-
 基于 Go 重写
 
 ## 提示

src/ZhiXueLite/app/admin/routes.py

@@ -5,7 +5,7 @@
 from sqlalchemy import select
 from app.database import db
 from app.database.models import Exam, School, ZhiXueStudentAccount, User
-from app.utils.paginate import paginated_json
+from app.utils.paginate import paginate_query
 from loguru import logger
 
 admin_bp = Blueprint("admin", __name__)
@@ -31,12 +31,10 @@ def list_schools():
     per_page = request.args.get("per_page", 10, type=int)
     query = request.args.get("query", "", type=str)
 
-    stmt = select(School)
+    stmt = select(School).order_by(School.id.asc())
     if query:
         stmt = stmt.where(School.name.contains(query) | School.id.contains(query))
-
-    schools = db.session.scalars(stmt).all()
-    paginated_schools = paginated_json(schools, page, per_page)
+    paginated_schools = paginate_query(stmt, page, per_page)
     school_list = [{"id": school.id, "name": school.name} for school in paginated_schools["items"]]
 
     return jsonify({
@@ -56,9 +54,7 @@ def list_users():
     stmt = select(User).order_by(User.id.asc())
     if query:
         stmt = stmt.where(User.username.contains(query))
-
-    users = db.session.scalars(stmt).all()
-    paginated_users = paginated_json(users, page, per_page)
+    paginated_users = paginate_query(stmt, page, per_page)
     user_list = [user.to_dict_all() for user in paginated_users["items"]]
 
     return jsonify({
@@ -75,12 +71,10 @@ def list_zhixue_accounts():
     per_page = request.args.get("per_page", 10, type=int)
     query = request.args.get("query", "", type=str)
 
-    stmt = select(ZhiXueStudentAccount)
+    stmt = select(ZhiXueStudentAccount).order_by(ZhiXueStudentAccount.id.asc())
     if query:
         stmt = stmt.where(ZhiXueStudentAccount.username.contains(query))
-
-    zhixue_accounts = db.session.scalars(stmt).all()
-    paginated_accounts = paginated_json(zhixue_accounts, page, per_page)
+    paginated_accounts = paginate_query(stmt, page, per_page)
     account_list = [account.to_dict_all() for account in paginated_accounts["items"]]
 
     return jsonify({
@@ -97,12 +91,10 @@ def list_exams():
     per_page = request.args.get("per_page", 10, type=int)
     query = request.args.get("query", "", type=str)
 
-    stmt = select(Exam).order_by(Exam.created_at.desc())
+    stmt = select(Exam).order_by(Exam.created_at.desc(), Exam.id.desc())
     if query:
         stmt = stmt.where(Exam.name.contains(query) | Exam.id.contains(query))
-
-    exams = db.session.scalars(stmt).all()
-    paginated_exams = paginated_json(exams, page, per_page)
+    paginated_exams = paginate_query(stmt, page, per_page)
     exam_list = [
         {
             "id": exam.id,

src/ZhiXueLite/app/exam/routes.py

@@ -5,13 +5,13 @@
 import os
 import time
 from openpyxl import Workbook
-from sqlalchemy import func, literal, select, desc, union_all
+from sqlalchemy import func, select
 from app.database import db
 from app.database.models import Exam, ExamSchool, PermissionLevel, Score, User, UserExam, ZhiXueTeacherAccount, PermissionType
 from app.models.exceptions import FailedToGetTeacherAccountError
 from app.models.teacher import login_teacher_session
 from app.task.repository import create_task
-from app.utils.paginate import paginated_json
+from app.utils.paginate import paginate_query
 from app import limiter
 from flask_limiter.util import get_remote_address
 
@@ -133,9 +133,8 @@ def get_exam_list():
     elif not (start_time == 0 and end_time == 0):
         return jsonify({"success": False, "message": "参数不合法"}), 400
 
-    exams = db.session.scalars(stmt.order_by(desc(Exam.created_at))).all()
-
-    paginated_exams = paginated_json(exams, page, per_page)
+    stmt = stmt.order_by(Exam.created_at.desc(), Exam.id.desc())
+    paginated_exams = paginate_query(stmt, page, per_page)
     exam_list = []
 
     has_global_permission = current_user.has_permission(

src/ZhiXueLite/app/task/routes.py

@@ -1,9 +1,9 @@
 from flask import Blueprint, request, jsonify
 from flask_login import login_required, current_user
-from sqlalchemy import select, desc
+from sqlalchemy import select
 from app.database import db
 from app.database.models import BackgroundTask, TaskStatus
-from app.utils.paginate import paginated_json
+from app.utils.paginate import paginate_query
 from . import repository as task_repo
 
 
@@ -49,9 +49,8 @@ def get_user_tasks():
         except ValueError:
             return jsonify({"success": False, "message": "无效的状态值"}), 400
 
-    tasks = db.session.scalars(stmt.order_by(desc(BackgroundTask.created_at))).all()
-
-    paginated_tasks = paginated_json(tasks, page, per_page)
+    stmt = stmt.order_by(BackgroundTask.created_at.desc(), BackgroundTask.id.desc())
+    paginated_tasks = paginate_query(stmt, page, per_page)
     task_list = [task.to_dict() for task in paginated_tasks["items"]]
 
     return jsonify({

src/ZhiXueLite/app/teacher/routes.py

@@ -1,12 +1,10 @@
-from functools import wraps
 from flask import Blueprint, jsonify, request
 from flask_login import current_user, login_required
-from loguru import logger
 from sqlalchemy import select
 from app.database import db
 from app.database.models import School, ZhiXueTeacherAccount
 from app.models.teacher import login_teacher
-from app.utils.paginate import paginated_json
+from app.utils.paginate import paginate_query
 
 teacher_bp = Blueprint("teacher", __name__)
 
@@ -33,9 +31,9 @@ def get_teacher_list():
         stmt = stmt.where(ZhiXueTeacherAccount.username.contains(query) |
                           School.name.contains(query))
 
-    teachers = db.session.scalars(stmt).all()
+    stmt = stmt.order_by(ZhiXueTeacherAccount.id.asc())
 
-    paginated_teachers = paginated_json(teachers, page, per_page)
+    paginated_teachers = paginate_query(stmt, page, per_page)
 
     teacher_list = [{
         "id": teacher.id,

src/ZhiXueLite/app/utils/paginate.py

@@ -1,28 +1,22 @@
-def paginate(items, page, per_page):
-    if not isinstance(items, list):
-        raise ValueError("Items must be a list.")
+from sqlalchemy import func, select
+from app.database import db
 
-    if page < 1 or per_page < 1:
-        raise ValueError("Page and per_page must be greater than 0.")
 
-    start = (page - 1) * per_page
-    end = start + per_page
-    paginated_items = items[start:end]
+def paginate_query(stmt, page, per_page):
+    """数据库级别分页，stmt 应已包含 where/order_by 条件"""
+    total = db.session.scalar(select(func.count()).select_from(stmt.subquery()))
 
-    return paginated_items, len(items), end
+    stmt = stmt.limit(per_page).offset((page - 1) * per_page)
+    items = db.session.scalars(stmt).all()
 
-
-def paginated_json(items, page, per_page):
-    """将数据转换为分页的 JSON 格式"""
-    paginated_items, total, end = paginate(items, page, per_page)
     return {
-        "items": paginated_items,
+        "items": items,
         "pagination": {
             "page": page,
             "per_page": per_page,
             "total": total,
             "pages": (total + per_page - 1) // per_page,
             "has_prev": page > 1,
-            "has_next": end < total
+            "has_next": page * per_page < total
         },
     }