Python itertools.batched แบบ strict: จับข้อมูลชุดสุดท้ายที่มาไม่ครบ

งานแบ่งข้อมูลทีละก้อนมีคำถามสำคัญหนึ่งข้อ: ก้อนสุดท้ายสั้นได้หรือถือว่าข้อมูลเสียครับ itertools.batched ตอบได้ทั้งสองแบบ โดยค่าปกติยอมให้ก้อนท้ายสั้น ส่วน strict=True จะยก ValueError เมื่อจำนวนสมาชิกหารขนาดก้อนไม่ลงตัว

WHY: ก้อนสั้นอาจเป็นข้อมูลปกติหรือสัญญาณอันตราย

ส่งอีเมลครั้งละร้อยรายการ ก้อนสุดท้ายมีสามสิบรายการเป็นเรื่องปกติ แต่ถ้าอ่านพิกัดทีละสามค่า หรือแบ่ง record แบบ fixed-width การเหลือหนึ่งหรือสองค่าแปลว่า input ถูกตัด การใช้ loop เดียวกันโดยไม่ประกาศ policy ทำให้ reviewer เดาเจตนา

เอกสาร Python ระบุว่า batched ใช้งานแบบ lazy: มันดึง input เท่าที่ต้องใช้เติมก้อน ไม่แปลง iterable ทั้งหมดเป็น list ก่อน จึงเหมาะกับ stream และข้อมูลขนาดใหญ่ แต่เมื่อเปิด strict ข้อผิดพลาดจะเกิดตอนเดินถึงก้อนสุดท้าย ไม่ใช่ตอนสร้าง iterator

HOW: ทำ Policy ให้เห็นใน Call Site

พารามิเตอร์ strict เริ่มใน Python 3.13 (แม้ batched มีตั้งแต่ 3.12) ตัวอย่างจึงทำ version boundary ให้มองเห็นและยังทดสอบ non-strict path บน 3.12 ได้จริง

import unittest
import sys
from itertools import batched


STRICT_BATCHING_AVAILABLE = sys.version_info >= (3, 13)


def triples(values):
    if not STRICT_BATCHING_AVAILABLE:
        raise RuntimeError("strict batching requires Python 3.13+")
    return list(batched(values, 3, strict=True))


class BatchedTests(unittest.TestCase):
    def test_runtime_requirement_is_explicit(self):
        if STRICT_BATCHING_AVAILABLE:
            self.assertEqual(triples([1, 2, 3]), [(1, 2, 3)])
        else:
            with self.assertRaisesRegex(RuntimeError, "Python 3.13"):
                triples([1, 2, 3])

    @unittest.skipUnless(
        STRICT_BATCHING_AVAILABLE,
        "strict=True requires Python 3.13+",
    )
    def test_complete_batches(self):
        self.assertEqual(
            triples([1, 2, 3, 4, 5, 6]),
            [(1, 2, 3), (4, 5, 6)],
        )

    @unittest.skipUnless(
        STRICT_BATCHING_AVAILABLE,
        "strict=True requires Python 3.13+",
    )
    def test_incomplete_batch_is_rejected(self):
        with self.assertRaises(ValueError):
            triples([1, 2, 3, 4])

    def test_non_strict_mode_keeps_remainder(self):
        self.assertEqual(
            list(batched("ABCDE", 2)),
            [("A", "B"), ("C", "D"), ("E",)],
        )


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

ตั้งชื่อ wrapper ตาม domain อย่าง triples ดีกว่าส่ง strict=True กระจายทุกที่ เพราะชื่อบอกว่าก้อนที่ไม่ครบผิดสัญญา และเปลี่ยน policy ภายหลังได้จากจุดเดียว

ระวัง Error ที่มาช้า

เพราะเป็น iterator การเขียน groups = batched(source, 3, strict=True) ยังไม่อ่าน source และยังไม่ยก error จนกว่าจะวนครบ หากต้อง validate ก่อนเริ่ม side effect อย่าส่งแต่ละก้อนไปปลายทางทันทีแล้วค่อยพบว่าก้อนสุดท้ายเสีย ให้ stage ลงพื้นที่ชั่วคราวหรือทำ validation pass เมื่อ input ย้อนอ่านได้

สำหรับ stream ให้ออกแบบ compensation หรือ idempotency เพราะ batch ก่อนหน้าอาจประมวลผลแล้ว ส่วน n ต้องมากกว่าศูนย์ มิฉะนั้น API ยก error ทันที

Recap

• ใช้โหมดปกติเมื่อ remainder เป็นข้อมูลถูกต้อง
• ใช้ strict=True เมื่อขนาดก้อนเป็นส่วนหนึ่งของ schema
• จำว่า iterator อ่านแบบ lazy และ error อาจเกิดท้าย stream
• ทดสอบทั้ง input ลงตัว, ไม่ลงตัว และว่าง

พรุ่งนี้เลือก loop ที่แบ่ง list เองหนึ่งจุด แล้วเขียนให้เห็นชัดว่าก้อนสุดท้ายสั้นได้หรือไม่ได้ครับ