SQLAlchemyモジュールを使用したページネーションの実装について説明します。

まず、paginationモジュールをインストールします。

pip install pagination

以下はpaginationモジュールに含まれるPaginationクラスのコードです。このコードは直接コピーして利用することも可能です。

def int_ceil(x, y):
    """
    math.ceil(x / y) と同等
    :param x:
    :param y:
    :return:
    """
    q, r = divmod(x, y)
    if r:
        q += 1
    return q

class Pagination(object):
    def __init__(self, query, page=1, per_page=10, per_nav=10,
                 map_=lambda x: x):
        self.first = 1
        self.total = query.count()
        if self.total == 0:
            self.last = 1
        else:
            self.last = int_ceil(self.total, per_page)
        self.page = max(min(self.last, page), 1)

        self.prev = max(self.page - 1, 1)
        self.has_prev = self.prev != self.page
        self.next = min(self.page + 1, self.last)
        self.has_next = self.next != self.page

        self.nav_head = per_nav * (int_ceil(self.page, per_nav) - 1) + 1
        self.nav_tail = min(self.last, self.nav_head + per_nav - 1)

        self.nav_prev = max(self.page - per_nav, 1)
        self.has_nav_prev = self.nav_prev < self.nav_head
        self.nav_next = min(self.page + per_nav, self.last)
        self.has_nav_next = self.nav_next > self.nav_tail

        self.pages = range(self.nav_head, self.nav_tail + 1)

        start = (self.page - 1) * per_page
        self.items = map(map_, query[start: start + per_page])

使用方法:

このページネーションクラスは、SQLAlchemyのqueryオブジェクトに対してページネーションを適用する場合に使用します。

from sqlalchemy_paginate import Pagination

# SQLAlchemyのqueryオブジェクトをPaginationクラスに渡します
query = db_session.query(YourTableName)
pagination_result = Pagination(query, page=3, per_page=2)

# pagination_result.items で指定したページに対応するデータオブジェクトのリストを取得できます
for obj in pagination_result.items:
    print(obj.name)

Flask-SQLAlchemy におけるページネーション

Flask-SQLAlchemy では、dbオブジェクトのpaginateメソッドを直接使用してページネーションを実装できます。これにより、クエリ結果とページネーションに関する各種情報を含むPaginationオブジェクトが生成されます。

以下は、BaseQueryのPaginationインスタンスから標準的なページネーション情報を取得するヘルパー関数です。

from flask_sqlalchemy import BaseQuery

def get_pagination_dict(pagination: BaseQuery.Pagination) -> dict | None:
    '''
    BaseQueryのPaginationインスタンスから標準的なページネーション情報を辞書形式で返します。
    '''
    if pagination is None:
        return None

    return {
        'has_next': pagination.has_next,
        'has_prev': pagination.has_prev,
        'page': pagination.page,
        'pages': pagination.pages,
        'total': pagination.total,
        'per_page': pagination.per_page,
        'page_item_num': len(pagination.items),
    }

以下は、ユーザーリストを取得し、ページネーションを適用するFlaskのルートハンドラ例です。

from flask import request, jsonify
from flask_sqlalchemy import BaseQuery
# 他に必要なインポート（AuthUser, AuthGroup, BizUnit, ConstBizUnitType, token_required_with_user, jsonify_return_body, formatter_account_info_in_list, generate_orderby_textなど）

@auth_blueprint.route('/list_accounts', methods=['POST'])
@token_required_with_user
def list_accounts(user):
    '''
    システムユーザーリスト取得API
    '''
    if user.group_id not in [0, 1]:
        return jsonify_return_body(['result', 'code', 'msg'], result=False, msg='権限がありません')

    filters = []
    pagination_params = {
        'page': 1,
        'per_page': 10
    }

    # デフォルトの並び順
    order_by_config = [
        {'key': 'auth_user.username', 'desc': True}
    ]

    # リクエストボディからページネーションと並び順のパラメータを取得
    if 'pageIndex' in request.json:
        pagination_params['page'] = request.json['pageIndex']
    if 'pageSize' in request.json:
        pagination_params['per_page'] = request.json['pageSize']
    if 'orderBy' in request.json:
        order_by_config = request.json['orderBy']

    # 並び順テキストの生成
    order_by_clause = generate_orderby_text(order_by_config)

    # ユーザーリストのクエリを構築
    query_users: BaseQuery = db_sqlalchemy.session.query(AuthUser, AuthGroup, BizUnit, ConstBizUnitType) \
        .select_from(AuthUser) \
        .outerjoin(AuthGroup, AuthUser.group_id == AuthGroup.id) \
        .outerjoin(BizUnit, AuthUser.biz_unit_id == BizUnit.id) \
        .outerjoin(ConstBizUnitType, BizUnit.type == ConstBizUnitType.id) \
        .filter(*filters) \
        .where(AuthUser.group_id != 0) \
        .order_by(order_by_clause)

    # paginateメソッドを使用してページネーションオブジェクトを生成
    pagination_obj = query_users.paginate(
        page=pagination_params['page'],
        per_page=pagination_params['per_page'],
        error_out=True,
        max_per_page=None
    )

    # 結果を整形して返す
    return jsonify_return_body(
        body_key_list=['code', 'msg', 'data', 'pagination'],
        result=True,
        data=[formatter_account_info_in_list(user, *i) for i in pagination_obj.items],
        pagination=get_pagination_dict(pagination_obj)
    )

query.paginate(...)メソッドを呼び出すことで、現在のページに対応するデータを含むPaginationオブジェクトが生成されます。このオブジェクトのitems属性には、そのページのデータリストが含まれます。

フィルタ条件の適用フロー

複数のフィルタ条件が存在する場合、FlaskのORMでは、queryオブジェクトがデータベースに実際にクエリを発行する前（first()やall()などのメソッドが呼び出される前）であれば、連続してfilter()メソッドを適用できます。各filter()呼び出しは、既存のフィルタ条件にAND条件で結合されます。つまり、すべてのフィルタ条件を満たすレコードのみが取得されます。

以下は、フィルタ条件の適用フローを示すテストコード例です。

@project_blueprint.route('/find_list111', methods=["GET", ])
@token_required
def find_list111():
    # 初期クエリ
    query = db_sqlalchemy.session.query(ProjectDtw)
    # SELECT id, name, create_by_user_id, img_url, status FROM project_dtw

    # 最初のフィルタ条件を追加 (id < 8)
    query = query.filter(ProjectDtw.id < 8)
    # SELECT id, name, create_by_user_id, img_url, status FROM project_dtw WHERE id < 8

    # 2番目のフィルタ条件を追加 (id > 2)
    query = query.filter(ProjectDtw.id > 2)
    # SELECT id, name, create_by_user_id, img_url, status FROM project_dtw WHERE id < 8 AND id > 2

    # ページネーションパラメータを取得
    page = int(request.args.get('page', 1))
    per_page = int(request.args.get('per_page', 20))

    # ページネーションを適用
    pagination = query.paginate(page=page, per_page=per_page, error_out=True, max_per_page=None)

    # ページネーション結果のアイテムを取得
    items_on_page = pagination.items

    # 結果を整形
    info_list = []
    for obj in items_on_page:
        info_list.append({
            'id': obj.id,
            'name': obj.name,
            'img_url': obj.img_url,
            'status': obj.status,
        })

    # JSONレスポンスを返す
    return jsonify_return_body(
        body_key_list=['code', 'msg', 'data', 'pagination'],
        result=True,
        data=info_list,
        pagination=get_pagination_dict(pagination) # 上記で定義したヘルパー関数を使用
    )

この例では、id < 8とid > 2という2つのフィルタ条件がANDで結合され、最終的なSQLクエリが生成されます。paginate()メソッドが呼び出された時点で、これらのフィルタ条件とページネーションが適用されたクエリが実行されます。

ユーザーリスト取得コード例

以下は、ユーザーリストを取得するAPIの実装例です。このAPIは、ページネーション、並び替え、および様々な条件（ユーザー名による部分一致、ロールID、グループID、複数のグループIDリスト）によるフィルタリングをサポートします。

from flask import request, jsonify
from sqlalchemy import func
from typing import List, Tuple
from sqlalchemy.orm import BaseQuery
# 他に必要なインポート（AuthUser, AuthRole, RsAuthUserGroup, AuthGroup, value_like_filter, in_list_filter, generate_orderby_text, get_pagination_record_from_request_and_query, formatter_account_info, blueprint, db, body_type_is_dictなど）

@blueprint.route('/user_find_list', methods=['POST'])
@body_type_is_dict
def user_find_list():
    """
    ユーザーリスト検索API
    """

    filters_list = [] # フィルタ条件リスト
    order_by_config = [{'key': 'auth_user.id', 'desc': True}] # デフォルトの並び順

    use_group_filter = False
    use_group_all_in_filter = False
    group_id_all_in_list = []

    # リクエストボディからパラメータを取得
    request_data = request.json

    if 'order_by' in request_data:
        order_by_config = request_data['order_by']

    if 'username' in request_data:
        filters_list.append(value_like_filter(AuthUser.username, request_data['username']))

    if 'role_id' in request_data:
        filters_list.append(AuthUser.role_id == request_data['role_id'])

    # 特定のグループIDでフィルタリング
    if 'group_id' in request_data:
        use_group_filter = True
        filters_list.append(AuthGroup.id == request_data['group_id'])

    # 指定されたグループIDのいずれかに属するユーザーでフィルタリング
    if 'group_id_list' in request_data:
        use_group_filter = True
        filters_list.append(in_list_filter(AuthGroup.id, request_data['group_id_list']))

    # 指定されたすべてのグループに属するユーザーでフィルタリング
    if 'group_id_all_in_list' in request_data:
        use_group_all_in_filter = True
        group_id_all_in_list = request_data['group_id_all_in_list']

    # 並び順テキストを生成
    order_by_clause = generate_orderby_text(order_by_config)

    query: BaseQuery

    if use_group_filter:
        # グループフィルタが有効な場合のクエリ
        query = db.session.query(AuthUser, AuthRole) \
            .select_from(AuthUser) \
            .outerjoin(AuthRole, AuthUser.role_id == AuthRole.id) \
            .outerjoin(RsAuthUserGroup, AuthUser.id == RsAuthUserGroup.user_id) \
            .outerjoin(AuthGroup, RsAuthUserGroup.group_id == AuthGroup.id) \
            .filter(*filters_list) \
            .order_by(order_by_clause)

    elif use_group_all_in_filter:
        # すべての指定グループに属するユーザーを検索するためのサブクエリ
        subquery = db.session.query(RsAuthUserGroup.user_id) \
            .filter(RsAuthUserGroup.group_id.in_(group_id_all_in_list)) \
            .group_by(RsAuthUserGroup.user_id) \
            .having(
                func.count(RsAuthUserGroup.group_id.in_(group_id_all_in_list)) == len(group_id_all_in_list)
            )

        # サブクエリを使用してユーザーオブジェクトを取得
        query = db.session.query(AuthUser, AuthRole) \
            .select_from(AuthUser) \
            .outerjoin(AuthRole, AuthUser.role_id == AuthRole.id) \
            .filter(AuthUser.id.in_(subquery))
    else:
        # 通常のフィルタリングクエリ
        query = db.session.query(AuthUser, AuthRole) \
            .select_from(AuthUser) \
            .outerjoin(AuthRole, AuthUser.role_id == AuthRole.id) \
            .filter(*filters_list) \
            .order_by(order_by_clause)

    # リクエストからページネーションとレコードを取得
    pagination, records = get_pagination_record_from_request_and_query(request, query, 1, 10)

    records: List[Tuple[AuthUser, AuthRole]] # 型ヒント

    user_info_in_list = []
    for user, role in records:
        user_info_in_list.append(formatter_account_info(user, role))

    # 結果をJSONで返す
    return jsonify_return_body(
        body_key_list=['code', 'msg', 'data'],
        result=True,
        data=user_info_in_list,
        pagination=pagination # ページネーション情報も返す
    )