functools, itertools, operator, heapq và bisect: những viên ngọc trong thư viện chuẩn

Rất nhiều đoạn code Python dài 10-20 dòng thực ra có thể viết bằng một dòng gọi hàm trong thư viện chuẩn - và thường chạy nhanh hơn vì phần lặp được viết bằng C. Bài này là một chuyến tham quan có chọn lọc những công cụ ứng dụng nhiều nhất trong functools, itertools, operator, heapq và bisect, kèm theo những bẫy mà tài liệu chính thức chỉ nhắc thoáng qua.

Trong bài này, bạn sẽ học:

• Cache kết quả hàm với cache/lru_cache - và vì sao dùng nó trên method có thể gây rò rỉ bộ nhớ
• partial, singledispatch, total_ordering, reduce
• Xử lý dữ liệu dạng luồng với itertools: chain, islice, groupby, batched, pairwise, accumulate, tổ hợp
• operator.itemgetter/attrgetter - nhanh và gọn hơn lambda
• heapq cho top-k và hàng đợi ưu tiên, bisect cho tìm kiếm trên dãy đã sắp xếp

functools

cache và lru_cache: ghi nhớ kết quả

Hàm Fibonacci đệ quy kinh điển gọi lại chính nó với cùng tham số rất nhiều lần:

import functools
import timeit

def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)

@functools.cache                 # Python 3.9+, tương đương lru_cache(maxsize=None)
def fib_cached(n):
    return n if n < 2 else fib_cached(n - 1) + fib_cached(n - 2)

print(timeit.timeit(lambda: fib(30), number=1))          # ~0.12s
print(timeit.timeit(lambda: fib_cached(30), number=1))   # ~0.000007s
print(fib_cached.cache_info())
# CacheInfo(hits=28, misses=31, maxsize=None, currsize=31)

Từ O(2ⁿ) xuống O(n) chỉ với một dòng decorator. Bên trong, cache là một dict với key được tạo từ các tham số - vì vậy:

1. Tham số phải hashable.

import functools

@functools.lru_cache
def total(items):
    return sum(items)

total((1, 2, 3))    # OK
total([1, 2, 3])    # TypeError: unhashable type: 'list'

2. f(1, 2) và f(a=1, b=2) là hai key khác nhau - cùng kết quả nhưng được cache hai lần.

3. cache không giới hạn kích thước. Với hàm nhận đầu vào từ người dùng (URL, ID…), cache sẽ lớn mãi - một dạng memory leak. Dùng lru_cache(maxsize=1024): khi đầy, bỏ phần tử lâu nhất chưa được dùng (Least Recently Used).

4. Kết quả được trả về chính là object đã cache. Nếu hàm trả về list và người gọi sửa list đó, lần gọi sau nhận list đã bị sửa. Hãy trả về dữ liệu bất biến (tuple, frozenset) từ hàm được cache.

Bẫy: lru_cache trên method

import functools, gc, weakref

class UserService:
    @functools.lru_cache(maxsize=None)
    def get_user(self, user_id):
        return {"id": user_id}

svc = UserService()
svc.get_user(1)

ref = weakref.ref(svc)
del svc
gc.collect()
print(ref() is not None)    # True - object KHÔNG được giải phóng!

self là một phần của key cache, và cache nằm trên hàm của class (sống mãi). Vì vậy mọi instance từng gọi method bị giữ sống vĩnh viễn. Cách xử lý:

• Nếu kết quả không phụ thuộc self → chuyển thành hàm cấp module hoặc @staticmethod có cache.
• Cache cho từng instance → tạo cache trong __init__: self.get_user = functools.lru_cache(maxsize=128)(self._get_user).
• Thuộc tính tính một lần → functools.cached_property (xem bài Descriptor), cache nằm trong __dict__ của instance nên được giải phóng cùng object.

partial: cố định sẵn một số tham số

from functools import partial

def power(base, exp):
    return base ** exp

square = partial(power, exp=2)
cube = partial(power, exp=3)
print(square(5), cube(2))       # 25 8

# Rất tiện khi API cần một hàm không tham số / ít tham số hơn
int_from_binary = partial(int, base=2)
print(int_from_binary("1011"))  # 11

Khác với lambda, partial pickle được (nếu hàm gốc pickle được) - dùng được với multiprocessing - và có thể xem lại bên trong: square.func, square.keywords.

singledispatch: “overload” theo kiểu của tham số đầu tiên

Thay vì một chuỗi if isinstance(...) elif isinstance(...):

from functools import singledispatch
from datetime import date
from decimal import Decimal

@singledispatch
def to_json(value):
    raise TypeError(f"không hỗ trợ kiểu {type(value).__name__}")

@to_json.register
def _(value: str):
    return f'"{value}"'

@to_json.register
def _(value: int | float):          # union type được hỗ trợ từ 3.11
    return str(value)

@to_json.register
def _(value: date):
    return f'"{value.isoformat()}"'

@to_json.register
def _(value: list):
    return "[" + ", ".join(to_json(v) for v in value) + "]"

print(to_json(["An", 16, date(2026, 9, 23)]))   # ["An", 16, "2026-09-23"]

Ưu điểm lớn: module khác có thể đăng ký thêm kiểu mà không sửa code gốc - to_json.register(Decimal, lambda v: str(v)). Cho method trong class, dùng singledispatchmethod.

total_ordering: viết 2 phương thức, có đủ 6 phép so sánh

from functools import total_ordering

@total_ordering
class Version:
    def __init__(self, s):
        self.parts = tuple(map(int, s.split(".")))

    def __eq__(self, other):
        return self.parts == other.parts

    def __lt__(self, other):
        return self.parts < other.parts

print(Version("3.10.1") > Version("3.9.7"))    # True (so sánh chuỗi sẽ cho False!)
print(max(["3.9", "3.10", "3.2"], key=Version)) # 3.10

total_ordering tự sinh <=, >, >= từ __eq__ và __lt__. (Nếu class là dataclass, @dataclass(order=True) cũng làm được điều này.)

reduce: gộp một dãy thành một giá trị

from functools import reduce
import operator

print(reduce(operator.mul, [1, 2, 3, 4, 5]))          # 120
merged = reduce(lambda a, b: a | b, [{"a": 1}, {"b": 2}, {"a": 3}])
print(merged)                                          # {'a': 3, 'b': 2}

Với các phép gộp phổ biến, hãy ưu tiên hàm có sẵn vì dễ đọc hơn: sum, math.prod, max, min, any, all, "".join.

itertools: xử lý dữ liệu như dòng chảy

Mọi hàm trong itertools trả về iterator lười - chỉ tính phần tử khi được yêu cầu. Bạn có thể xử lý file hàng GB hoặc dãy vô hạn mà không cần nạp tất cả vào RAM (xem thêm Generators và Iterators).

Nối và cắt

from itertools import chain, islice

lists = [[1, 2], [3], [4, 5, 6]]
print(list(chain.from_iterable(lists)))     # [1, 2, 3, 4, 5, 6] - làm phẳng 1 cấp

# islice: cắt iterator như slice, không cần len()
with open(__file__) as f:
    first_3_lines = list(islice(f, 3))       # đọc đúng 3 dòng, không đọc hết file

batched (3.12+): chia thành từng lô

Rất hay dùng khi gửi dữ liệu lên API hoặc insert database theo lô:

from itertools import batched

ids = range(1, 8)
for batch in batched(ids, 3):
    print(batch)
# (1, 2, 3)
# (4, 5, 6)
# (7,)

Từ 3.13, batched(..., strict=True) ném ValueError nếu lô cuối không đủ.

pairwise (3.10+): các cặp liền kề

from itertools import pairwise

prices = [100, 104, 101, 110]
changes = [b - a for a, b in pairwise(prices)]
print(changes)      # [4, -3, 9]

accumulate: tổng dồn và hơn thế

from itertools import accumulate

print(list(accumulate([1, 2, 3, 4])))          # [1, 3, 6, 10]   tổng tiền tố
print(list(accumulate([3, 1, 4, 1, 5], max)))  # [3, 3, 4, 4, 5] giá trị lớn nhất tới hiện tại

Tổng tiền tố (prefix sum) là kỹ thuật cơ bản để trả lời “tổng đoạn từ i đến j” trong O(1) - rất hay gặp trong bài thi lập trình.

groupby: gom nhóm các phần tử liền kề

from itertools import groupby

print([(k, list(g)) for k, g in groupby("aabba")])
# [('a', ['a', 'a']), ('b', ['b', 'b']), ('a', ['a'])]   <- 'a' xuất hiện HAI nhóm!

groupby chỉ gom các phần tử đứng cạnh nhau - khác với GROUP BY của SQL. Muốn gom theo khoá trên toàn bộ dữ liệu, phải sắp xếp theo cùng khoá trước:

from itertools import groupby
from operator import itemgetter

orders = [
    {"customer": "An", "total": 100},
    {"customer": "Bình", "total": 50},
    {"customer": "An", "total": 70},
]

key = itemgetter("customer")
for customer, group in groupby(sorted(orders, key=key), key=key):
    print(customer, sum(o["total"] for o in group))
# An 170
# Bình 50

Thêm một bẫy: mỗi group là iterator dùng chung dữ liệu với groupby; khi chuyển sang nhóm tiếp theo, nhóm trước không còn đọc được. Cần giữ lại thì chuyển thành list ngay (list(group)). Nếu dữ liệu không có thứ tự sẵn, defaultdict(list) thường đơn giản hơn.

Tổ hợp

from itertools import product, permutations, combinations

print(list(product("AB", [1, 2])))        # [('A', 1), ('A', 2), ('B', 1), ('B', 2)]
print(list(permutations("ABC", 2)))       # 6 hoán vị chập 2
print(list(combinations([1, 2, 3, 4], 2)))# 6 tổ hợp chập 2

# product thay cho vòng lặp lồng nhau
for x, y, z in product(range(2), repeat=3):
    pass                                  # 8 bộ (x, y, z) nhị phân

Các hàm khác đáng nhớ

Hàm	Công dụng
count(start, step)	đếm vô hạn - kết hợp zip để đánh số
cycle(it)	lặp vòng vô hạn (xoay vòng server, màu sắc)
repeat(x, n)	lặp một giá trị
zip_longest(a, b, fillvalue=...)	zip không cắt theo dãy ngắn nhất
takewhile(pred, it) / dropwhile	lấy/bỏ phần tử khi điều kiện còn đúng
compress(data, selectors)	lọc theo một dãy boolean
starmap(f, pairs)	f(*pair) cho từng cặp
tee(it, n)	tách một iterator thành n iterator độc lập

operator: thay lambda bằng hàm C

import operator
import timeit

data = [{"n": i % 100, "s": str(i)} for i in range(100_000)]

print(timeit.timeit(lambda: sorted(data, key=lambda d: d["n"]), number=10))         # ~0.061s
print(timeit.timeit(lambda: sorted(data, key=operator.itemgetter("n")), number=10)) # ~0.035s

itemgetter nhanh gần gấp đôi lambda tương đương vì không phải tạo frame Python cho mỗi lần gọi. Và nó còn tiện hơn:

from operator import itemgetter, attrgetter, methodcaller

rows = [("An", 20, 8.5), ("Bình", 19, 9.0), ("Chi", 20, 7.5)]
print(sorted(rows, key=itemgetter(1, 2)))       # sắp xếp theo tuổi, rồi theo điểm

get_name_age = itemgetter(0, 1)
print(get_name_age(rows[0]))                    # ('An', 20)

# attrgetter hỗ trợ đường dẫn có dấu chấm: attrgetter("address.city")
# methodcaller("strip") tương đương lambda s: s.strip()
print(list(map(methodcaller("upper"), ["a", "b"])))   # ['A', 'B']

Các toán tử cũng có dạng hàm: operator.add, operator.mul, operator.or_, operator.not_…

heapq: top-k và hàng đợi ưu tiên

heapq biến một list thường thành min-heap: phần tử nhỏ nhất luôn ở heap[0], thêm/lấy ra trong O(log n).

Top-k mà không cần sắp xếp toàn bộ

import heapq
import random

scores = [random.randint(0, 1_000_000) for _ in range(1_000_000)]

top5 = heapq.nlargest(5, scores)        # O(n log k) thay vì O(n log n)
print(top5)

students = [{"name": "An", "gpa": 3.2}, {"name": "Bình", "gpa": 3.8}, {"name": "Chi", "gpa": 3.5}]
print(heapq.nsmallest(2, students, key=lambda s: s["gpa"]))

Quy tắc: k nhỏ → nlargest/nsmallest; k = 1 → max/min; k gần bằng n → sorted(...)[:k].

Hàng đợi ưu tiên

import heapq
import itertools

class PriorityQueue:
    def __init__(self):
        self._heap = []
        self._counter = itertools.count()     # phá hoà để giữ thứ tự FIFO, tránh so sánh item

    def push(self, item, priority):
        heapq.heappush(self._heap, (priority, next(self._counter), item))

    def pop(self):
        priority, _, item = heapq.heappop(self._heap)
        return item

    def __len__(self):
        return len(self._heap)

pq = PriorityQueue()
pq.push("gửi email", priority=3)
pq.push("xử lý thanh toán", priority=1)
pq.push("ghi log", priority=5)
while pq:
    print(pq.pop())
# xử lý thanh toán
# gửi email
# ghi log

Bộ đếm _counter quan trọng: nếu hai task cùng priority, tuple sẽ so sánh tiếp tới item - với dict hay object không so sánh được, bạn sẽ gặp TypeError. Đây là mẫu dùng trong thuật toán Dijkstra, lập lịch công việc, merge nhiều file đã sắp xếp (heapq.merge).

Cần max-heap? Đẩy vào giá trị âm (-priority). (Python 3.14 bổ sung các hàm max-heap như heapq.heappush_max.)

bisect: tìm kiếm nhị phân trên dãy đã sắp xếp

import bisect

# Xếp loại theo điểm: tìm vị trí điểm trong các mốc
breakpoints = [5.0, 6.5, 8.0, 9.0]
grades = ["Yếu", "Trung bình", "Khá", "Giỏi", "Xuất sắc"]

def classify(score):
    return grades[bisect.bisect_right(breakpoints, score)]

print([classify(s) for s in [4.0, 6.5, 7.9, 8.0, 9.5]])
# ['Yếu', 'Khá', 'Khá', 'Giỏi', 'Xuất sắc']

import bisect

# Chèn mà vẫn giữ thứ tự (tìm vị trí O(log n), chèn O(n))
events = [10, 20, 40]
bisect.insort(events, 30)
print(events)      # [10, 20, 30, 40]

# Đếm số phần tử trong khoảng [lo, hi] trên dãy đã sắp xếp: O(log n)
def count_between(sorted_list, lo, hi):
    return bisect.bisect_right(sorted_list, hi) - bisect.bisect_left(sorted_list, lo)

print(count_between([1, 3, 3, 5, 7, 9], 3, 7))   # 4

Từ Python 3.10, các hàm bisect nhận tham số key=, tiện khi tìm trong list các object đã sắp xếp theo một trường.

Bài tập

1. Dùng lru_cache giải bài toán “số cách leo cầu thang n bậc, mỗi lần 1, 2 hoặc 3 bậc” với n = 500. Xem cache_info().
2. Cho file log dạng thời_gian,level,message (đã sắp xếp theo thời gian). Dùng groupby đếm số dòng mỗi level trong từng phút.
3. Dùng heapq.merge gộp 3 file số đã sắp xếp (mỗi file vài triệu dòng) thành một file đã sắp xếp mà không nạp hết vào RAM.
4. Dùng batched và ThreadPoolExecutor gửi 10.000 bản ghi lên một API giả lập, mỗi request 100 bản ghi.

Kết luận

• cache/lru_cache tăng tốc thần kỳ nhưng cần tham số hashable, có giới hạn kích thước, và tránh dùng trực tiếp trên method.
• partial, singledispatch, total_ordering giúp code ngắn và dễ mở rộng.
• itertools xử lý dữ liệu lười: batched, pairwise, accumulate, chain - nhớ groupby chỉ gom phần tử liền kề.
• itemgetter/attrgetter nhanh hơn và rõ nghĩa hơn lambda làm key.
• heapq cho top-k và hàng đợi ưu tiên; bisect cho tìm kiếm và chèn trên dãy đã sắp xếp.

Bài tiếp theo: Đo và tối ưu hiệu năng.