صفحه‌بندی سریع پایگاه داده با نشانگر

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

CREATE TABLE events (
  id INTEGER PRIMARY KEY,
  sequence INTEGER NOT NULL,
  label TEXT NOT NULL
);
CREATE INDEX events_cursor_idx ON events(sequence, id);

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

import base64
import json


def encode_cursor(row):
    data = json.dumps([row[1], row[0]], separators=(',', ':')).encode()
    return base64.urlsafe_b64encode(data).decode()


def decode_cursor(cursor):
    if len(cursor) > 128:
        raise ValueError('Cursor too long')
    try:
        values = json.loads(base64.b64decode(cursor, altchars=b'-_', validate=True))
    except (ValueError, TypeError) as error:
        raise ValueError('Invalid cursor') from error
    if not isinstance(values, list) or len(values) != 2 or any(type(v) is not int or v < 0 for v in values):
        raise ValueError('Cursor requires two nonnegative integers')
    return values


def page(connection, after=None, size=20):
    if type(size) is not int or not 1 <= size <= 100:
        raise ValueError('Page size must be 1..100')
    params = []
    where = ''
    if after is not None:
        where = 'WHERE (sequence, id) > (?, ?)'
        params.extend(decode_cursor(after))
    rows = connection.execute(f'SELECT id, sequence, label FROM events {where} ORDER BY sequence, id LIMIT ?', [*params, size]).fetchall()
    return {'rows': rows, 'next': encode_cursor(rows[-1]) if rows else None}

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

from pathlib import Path
import sqlite3
from pages import page

if __name__ == '__main__':
    with sqlite3.connect(':memory:') as connection:
        connection.executescript(Path(__file__).with_name('schema.sql').read_text())
        connection.executemany('INSERT INTO events VALUES (?, ?, ?)', [(i, i // 3, f'event-{i}') for i in range(1, 13)])
        cursor = None
        while True:
            result = page(connection, cursor, size=4)
            if not result['rows']:
                break
            print(result)
            cursor = result['next']

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

from pathlib import Path
import sqlite3
import unittest
from pages import decode_cursor, page


class PaginationTests(unittest.TestCase):
    def test_duplicate_sort_keys_and_concurrent_insert(self):
        with sqlite3.connect(':memory:') as connection:
            connection.executescript(Path(__file__).with_name('schema.sql').read_text())
            connection.executemany('INSERT INTO events VALUES (?, 1, ?)', [(i, str(i)) for i in range(1, 7)])
            first = page(connection, size=3)
            connection.execute("INSERT INTO events VALUES (0, 0, 'earlier')")
            second = page(connection, first['next'], size=3)
            self.assertEqual([r[0] for r in first['rows'] + second['rows']], list(range(1, 7)))
            self.assertEqual(page(connection, second['next'])['rows'], [])

    def test_invalid_cursor_and_size(self):
        with self.assertRaises(ValueError):
            decode_cursor('not-json')
        with self.assertRaises(ValueError):
            page(None, size=0)


if __name__ == '__main__':
    unittest.main()

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.