删除某些数据后，如何从SQLITE3表中获取前 10 个和后 10 个不同的group_ids

问：

从 sqlite3 表列中，我想在删除一些数据后以给定的偏移量提取前 10 个不同的值。

下面是一个示例脚本。

示例脚本：

import sqlite3
import random
from itertools import count
from typing import Literal

class Data:

    def __init__(self):
        self.con = sqlite3.connect("data.db",
                                   detect_types=sqlite3.PARSE_DECLTYPES,
                                   )
        self.cur = self.con.cursor()
        self.create_table()

    def create_table(self):
        table = """CREATE TABLE IF NOT EXISTS
        datatable(
            sn INTEGER,
            item_id TEXT PRIMARY KEY,
            group_id TEXT)
        """
        self.cur.execute("""DROP TABLE IF EXISTS datatable""")
        self.cur.execute(table)
        self.con.commit()

    def insert_data_row(self, data):
        sql = """INSERT OR IGNORE INTO datatable VALUES (?,?,?)"""
        self.cur.execute(sql, data)
        self.con.commit()


    def close(self):
        # 5. Close cursor & connection
        self.cur.close()
        self.con.close()

    def delete_group_id(self, group_id: str):
        sql = """DELETE FROM datatable WHERE group_id in (?)"""
        self.cur.execute(sql, (group_id,))
        self.con.commit()

    def get_previous_page_of_group_ids(self, group_id: str, span: int):
        sql1 = """SELECT MIN(sn) FROM 
                (SELECT sn from datatable WHERE group_id == (?))"""
        sql2 = """SELECT DISTINCT group_id FROM datatable 
                    WHERE sn < (?) ORDER BY group_id DESC LIMIT (?)     
                    OFFSET (?)"""
        self.cur.execute(sql1, (group_id,))
        start_sn = self.cur.fetchone()[0]
        print(f"{start_sn=}")
        self.cur.execute(sql2, (start_sn, span, 10))
        return self.cur.fetchall()

    def get_next_page_of_group_ids(self, group_id: str, span: int):
        sql1 = """SELECT MAX(sn) FROM 
                (SELECT sn from datatable WHERE group_id == (?))"""
        sql2 = """SELECT DISTINCT group_id FROM datatable
                    WHERE sn > (?) LIMIT (?)"""
        self.cur.execute(sql1, (group_id,))
        start_sn = self.cur.fetchone()[0]
        print(f"{start_sn=}")
        self.cur.execute(sql2, (start_sn, span,))
        return self.cur.fetchall()



if __name__ == "__main__":
    db = Data()
    db.create_table()
    counter = count()
    for i in range(50):
        repeats = random.randint(0, 5)
        for r in range(repeats):
            sn = next(counter)
            gid = f"G{i}"
            id = f"{gid}_F{r}"
            print(f"{sn=} {gid=} {id}")
            db.insert_data_row((sn, id, gid))
    start_gid = "G30"
    db.delete_group_id("G0")
    db.delete_group_id("G1")
    db.delete_group_id("G29")
    db.delete_group_id("G28")
    print(f"{start_gid=}")
    pb = db.get_previous_page_of_group_ids(start_gid, 10)
    print(f"{pb=}")
    nb = db.get_next_page_of_group_ids(start_gid, 10)
    print(f"{nb=}")
    db.close()

返回的答案是：

start_gid='G30'
start_sn=68
pb=[('G21',), ('G2',), ('G17',), ('G16',), ('G14',), ('G12',), ('G11',), ('G10',)]
start_sn=68
nb=[('G31',), ('G33',), ('G34',), ('G35',), ('G36',), ('G40',), ('G41',), ('G42',), ('G43',), ('G45',)]

pb并且似乎错了。它应该来自 和 。我无法弄清楚和方法中的错误。感谢您帮助理解此方法中的错误。谢谢。np((G27'),... to ..., (G18')((G31'),... to ..., (G40').get_previous_page_of_group_ids().get_next_page_of_group_ids()

更新：

我使用 SQLITE3 和 Python 命令的组合进行了重写。我可以知道编写此方法的纯SQLITE3方法吗？.get_previous_page_of_group_ids()

   def get_previous_page_of_group_ids(self, group_id: str, span: int):
        sql1 = """SELECT MIN(sn) FROM 
                (SELECT sn from datatable WHERE group_id == (?))"""
        sql2 = """SELECT group_id from datatable 
                WHERE sn < (?)"""
        self.cur.execute(sql1, (group_id,))
        start_sn = self.cur.fetchone()[0]
        print(f"{start_sn=}")
        self.cur.execute(sql2, (start_sn,))
        indexes = {int(i[0][1:]) for i in self.cur.fetchall()}
        print(f"{indexes=}")
        gids = [f"G{i}" for i in sorted(indexes, reverse=True)[:span]]
        gids.reverse()
        print(f"{gids=}")
        return gids

我还发现了一个问题，其中有时会导致方法和的结果出现错误。本声明__main__.get_next_page_of_group_ids().get_previous_page_of_group_ids()

repeats = random.randint(0, 5)

应该改为

repeats = random.randint(1, 5)