Түвшин 2-ын бүтээл: Хэмжигч хэрэгслийн сан

Танилцуулга

Түвшин 2-ын төгсгөлд хүрлээ. Арван нэгэн хичээлийн туршид бид функцийн дотоод амьдралыг бүрэн нээсэн. LEGB дүрмээс эхэлж, closure-ийн үүргэвчийг нээж, nonlocal-оор хайрцгийн агуулгыг өөрчилсөн. Decorator-ыг closure дээр дахин бүтээж, wraps-ээр мэргэжлийн түвшинд хүргэж, argument-тай гурван давхар бүтцийг эзэмшиж, давхарлалтын дарааллыг ойлгосон. Эцэст нь with-ийн протоколыг нээж, context manager-ийг class-аар ба generator-оор бичсэн. Энэ хичээлд тэр бүх мэдлэг нэг бодит бүтээгдэхүүнд нэгдэнэ.

Бид жижиг хэрэгслийн сан бүтээнэ: measurekit. Энэ нь гурван хэрэгсэлтэй байна. timer бол дурын кодын блокийг хэмжих context manager. timed бол бүхэл функцийг хэмжих decorator, @timed ба @timed(label="...") хоёр хэлбэрээр хоёуланд нь ажилладаг. retry бол алдаа гарвал функцийг дахин оролддог argument-тай decorator. Гурвуулаа wraps ашиглана, гурвуулаа pytest тесттэй, ажил Git branch дээр хийгдэнэ.

Өмнөх хичээлүүдээс ялгаатай нь энэ удаа бид "сан" бүтээж байна: өөр хүмүүс, эсвэл ирээдүйн өөрөө, өөр төслүүдэд импортлон хэрэглэх код. Энэ нь хэд хэдэн дизайны шийдвэр шаардана: хэрэглэгчид мэдээллийг хаашаа гаргахаа сонгох эрх өгөх, буруу тохиргоог эрт илрүүлэх, тестэд хэрэгслүүдийг хурдан, найдвартай шалгах боломж олгох. Эдгээр шийдвэр бүр өмнөх хичээлийн аль нэг санаанаас шууд урган гарна. Энэ бүтээлийн төгсгөлд @ ба with тэмдэг таных болно: та тэдгээрийг хэрэглэдэг төдийгүй, бусдад зориулан бүтээдэг болно.

Бид юу бүтээх вэ

Төслийн бүтэц гурван файлтай:

measure-tools/
    measurekit.py         хэрэгслийн сан
    test_measurekit.py    pytest тестүүд
    demo.py               гурван хэрэгслийг хамт харуулах скрипт

Кодыг бичихээс өмнө гурван дизайны шийдвэрийг тайлбарлая. Тэдгээр нь кодын бүх хэсэгт харагдана.

Мэдээллийг хаашаа гаргах вэ. Хичээл 5.1 ба 6.2-ын Timer-ууд үр дүнгээ print()-ээр хэвлэдэг байсан. Сангийн хувьд энэ нь хатуу шийдвэр: нэг хэрэглэгч хэвлэлт хүсэх, нөгөө нь лог файлд бичихийг хүсэх, тест бүр огт гаралтгүй байхыг хүснэ. Тиймээс бидний гурван хэрэгсэл бүгд report гэсэн тохиргоотой байна: мэдээллийг хүлээн авах функц. Анхдагч утга нь print. Хэрэглэгч өөр функц өгч болно, жишээлбэл list-ийн append. Энэ нь Хичээл 3.1-ээс хойшхи "функц бол утга" санааны бодит хэрэглээ.

Буруу тохиргоог эрт илрүүлэх. retry(times=0) утгагүй: функц нэг ч удаа дуудагдахгүй. Энэ алдааг функцийг анх дуудах үед биш, тохиргооны цагт, @ мөрөнд илрүүлэх нь зөв. Хичээл 4.3-т бид factory тохиргоог үүсэх мөчид шалгаж болно гэж тэмдэглэсэн, одоо үүнийг хэрэглэнэ.

Тестэд хүлээлтгүй байх. retry оролдлогын хооронд time.sleep()-ээр хүлээдэг. Хэрэв тест бүр жинхэнээр хүлээвэл тестийн багц удаан болно. Тиймээс retry-д sleep гэсэн тохиргоо нэмнэ, анхдагч нь time.sleep. Тест жинхэнэ хүлээлтийн оронд хүлээх хүсэлтийг list-д бичдэг функц өгнө. report-тай ижил санаа: хэрэгслийн гадаад хамаарлыг тохиргоогоор өгөх.

Алхам 0: branch ба хавтас

Энэ сургалтын repository-доо шинэ branch үүсгэнэ:

git switch -c level2-measurekit

Repository дотор measure-tools нэртэй хавтас үүсгэж, terminal-аар түүн рүү орно. venv идэвхтэй эсэхийг шалгаарай. Энэ бүтээл стандарт сан ба pytest-ээс өөр юу ч шаардахгүй.

Алхам 1: timer context manager

measurekit.py файл үүсгэж, эхний хэрэгслийг бичнэ:

python

"""Хугацаа хэмжих ба дахин оролдох жижиг хэрэгслүүд."""

import time
from contextlib import contextmanager
from functools import partial, wraps


@contextmanager
def timer(label="Блок", report=print):
    """with блокийн ажиллах хугацааг хэмжиж, report-оор мэдээлнэ."""
    result = {"label": label, "elapsed": None, "ok": False}
    start = time.perf_counter()
    try:
        yield result
        result["ok"] = True
    finally:
        result["elapsed"] = time.perf_counter() - start
        status = "" if result["ok"] else " (алдаатай дууслаа)"
        report(f"{label}: {result['elapsed']:.4f} секунд{status}")

Энэ бол Хичээл 6.3-ын Жишээ 1-ийн timer, гурван сайжруулалттай. Файлын эхэнд docstring бий: модулийн тайлбар, help(measurekit)-д харагдана. Импортын мөрөнд partial ба wraps бий, тэд дараагийн алхмуудад хэрэгтэй.

Нэгдүгээр сайжруулалт: report тохиргоо. Хэвлэх биш, report функцийг дуудна. Хоёрдугаар сайжруулалт: result dict-ийн “ok” түлхүүр. yield-ийн дараа, гэхдээ try дотор, result["ok"] = True мөр бий. Хичээл 6.3-ын trace-ийг санаарай: блок дотор алдаа гарвал тэр yield мөрөн дээр шидэгдэнэ. Тэгэхээр алдаатай үед энэ мөр алгасагдаж, ok нь False хэвээр үлдэнэ. Хэвийн үед yield хэвийн буцаж, мөр ажиллана. Энэ нь __exit__-ийн exc_type-гүйгээр блок амжилттай байсан эсэхийг мэдэх энгийн арга. Гуравдугаар сайжруулалт: тэр мэдээллээр finally дотор “(алдаатай дууслаа)” гэсэн тэмдэглэл нэмэгдэнэ, Хичээл 6.2-ын Timer class-ийн адил.

Шалгах цэг 1. measure-tools хавтаснаас Python-ий interactive горимыг нээнэ:

python

>>> from measurekit import timer
>>> with timer("Нийлбэр") as t:
...     total = sum(range(5_000_000))
...
Нийлбэр: 0.0734 секунд
>>> t
{'label': 'Нийлбэр', 'elapsed': 0.07341126, 'ok': True}
>>> messages = []
>>> with timer("Хуваах", report=messages.append):
...     1 / 0
...
Traceback (most recent call last):
  ...
ZeroDivisionError: division by zero
>>> messages
['Хуваах: 0.0000 секунд (алдаатай дууслаа)']

Хэвийн тохиолдолд хэмжилт хэвлэгдэж, t dict бүрэн дүүрсэн. Хоёр дахь тохиолдолд report-д messages.append өгсөн тул юу ч хэвлэгдээгүй, мэдээ list-д орсон. Алдаа гадагш гарсан ч finally ажиллаж, “(алдаатай дууслаа)” тэмдэглэлтэй мэдээ бичигдсэн. Тоонууд таны компьютер дээр өөр байна.

Алхам 2: timed decorator, хоёр хэлбэрээр

Одоо хамгийн сонирхолтой хэрэгсэл. Бид timed-ийг хоёр янзаар хэрэглэж болдог байлгамаар байна:

python

@timed
def build(): ...

@timed(label="Тайлан", report=logger)
def build(): ...

Эхнийх нь Хичээл 5.1-ийн энгийн decorator хэлбэр. Хоёр дахь нь Хичээл 5.3-ын argument-тай хэлбэр. Хичээл 5.3-ын Түгээмэл алдаанууд хэсэгт эдгээр хоёр хэлбэрийг хольж хэрэглэхэд алдаа гардгийг харсан. Гэвч нэг жижиг заль хоёуланг нэг функцэд дэмжих боломж олгоно. measurekit.py-ийн төгсгөлд нэмнэ:

python

def timed(func=None, *, label=None, report=print):
    """Функцийн дуудлага бүрийн хугацааг хэмждэг decorator."""
    if func is None:
        return partial(timed, label=label, report=report)
    if not callable(func):
        raise TypeError("timed: шошгыг label=... гэж түлхүүр үгээр өгнө үү")

    name = label or func.__name__

    @wraps(func)
    def wrapper(*args, **kwargs):
        with timer(name, report=report):
            return func(*args, **kwargs)

    return wrapper

Гарын үсгийг уншъя: func=None, *, label=None, report=print. Дунд нь ганц * тэмдэг бий. Энэ нь Python-д “*-ийн дараах бүх argument-ыг зөвхөн нэрээр нь өгнө” гэж хэлж байна. Тэгэхээр label ба report-ийг байрлалаар өгөх боломжгүй, зөвхөн label="..." гэж. Энэ шийдвэрийн учрыг удахгүй харна.

Функцийн бие гурван хэсэгтэй. Эхний if нь хоёр хэлбэрийг ялгадаг заль. Хоёр дахь if нь буруу хэрэглээг барина. Гурав дахь хэсэг нь танил decorator: wraps-тай wrapper. wrapper-ийн дотор Алхам 1-ийн timer-ийг хэрэглэсэн: функцийн дуудлагыг with блок дотор хийж, return-ийг шууд with дотроос хийсэн. Хичээл 6.1-ийн Жишээ 1-ийг санаарай: return with блокийг орхиход __exit__ заавал ажилладаг. Тэгэхээр хэмжилт хэвлэгдэж, үр дүн буцна. timed хэмжилтийн логикийг дахин бичээгүй, timer-ийг дахин ашигласан. Хэрэв хэзээ нэгэн цагт хэмжилтийн форматыг өөрчлөх шаардлага гарвал ганц газарт засна. label or func.__name__ нь label өгөгдөөгүй бол функцийн нэрийг шошго болгоно.

Trace: хоёр зам

Эхний if-ийн ажиллагааг хоёр хэлбэрээр мөрдье.

Зам А: @timed хаалтгүй.

Алхам А1. Хичээл 5.1-ийн томьёогоор build = timed(build). timed-д байрлалаар нэг argument ирсэн: build функц. func = build, label = None, report = print.

Алхам А2. func is None худал. callable(func) үнэн, учир нь функц дуудагдаж чадна. callable() бол Python-ий суурь функц: объектыг дуудаж болох эсэхийг хэлдэг.

Алхам А3. name = “build”. wrapper үүсч, үүргэвчинд func, name, report. wraps шошгыг солино. wrapper буцаж, build нэрэнд оноогдоно. Энэ нь Хичээл 5.2-ын ердийн decorator.

Зам Б: @timed(label="Тайлан") хаалттай.

Алхам Б1. Хичээл 5.3-ын томьёогоор build = timed(label="Тайлан")(build). Эхлээд timed(label="Тайлан") дуудагдана. func-д юу ч өгөгдөөгүй тул анхдагч None. label = "Тайлан".

Алхам Б2. func is None үнэн. timed partial(timed, label="Тайлан", report=print) буцаана. Хичээл 3.3-ын partial: timed функцийг label ба report бэхэлсэн хувилбар. Энэ бол тохиргооны цаг: decorator үйлдвэрлэгдлээ.

Алхам Б3. Тэр partial build-ээр дуудагдана: partial(...)(build). partial бэхэлсэн argument-уудыг нэмж, timed(build, label="Тайлан", report=print) гэж дуудна.

Алхам Б4. Одоо timed дахин ажиллана, гэхдээ func = build. Зам А-ийн Алхам А2-А3 руу шилжинэ: callable шалгалт, name = "Тайлан", wrapper үүсч буцна.

Тэгэхээр timed нэг функц атлаа хоёр үүрэгтэй: func ирвэл decorator, func ирэхгүй бол decorator үйлдвэрлэгч. Хичээл 5.3-ын гурван давхар бүтцийн гадна давхаргыг partial орлож байна: тусдаа decorator функц бичихийн оронд timed өөрийгөө тохиргоотойгоор бэхэлж буцаадаг. Энэ загвар бодит сангуудад түгээмэл бөгөөд та одоо түүний ард юу байгааг бүрэн мэдэж байна.

* тэмдэг яагаад хэрэгтэй вэ? Хэрэв label байрлалаар өгөгдөж болдог байсан бол @timed("Тайлан") гэж бичих уруу таталт гарна. Тэр үед "Тайлан" string func-д очиж, timed түүнийг функц гэж бодож боох гэж оролдоно. Хоёр дахь if яг энэ тохиолдлыг барьдаг: string дуудагдаж чадахгүй тул callable(func) худал, бид тодорхой мессежтэй TypeError шиднэ. Энэ хамгаалалтгүйгээр хэрэглэгч хачин, ойлгомжгүй алдаа авах байсан. Үүнийг Түгээмэл алдаанууд хэсэгт харна.

Шалгах цэг 2. Interactive горимыг дахин нээгээд (measurekit өөрчлөгдсөн):

python

>>> from measurekit import timed
>>> @timed
... def build_squares(n):
...     """n хүртэлх квадратуудын жагсаалт."""
...     return [x * x for x in range(n)]
...
>>> len(build_squares(1_000_000))
build_squares: 0.0419 секунд
1000000
>>> build_squares.__name__, build_squares.__doc__
('build_squares', 'n хүртэлх квадратуудын жагсаалт.')
>>> @timed(label="Нийлбэр бодох")
... def total(n):
...     return sum(range(n))
...
>>> total(10_000_000)
Нийлбэр бодох: 0.1481 секунд
49999995000000

Хоёр хэлбэр хоёулаа ажиллаж байна, wraps нэр ба тайлбарыг хадгалсан.

Алхам 3: retry decorator

measurekit.py-ийн төгсгөлд гурав дахь хэрэгслийг нэмнэ:

python

def retry(
    times=3,
    delay=0.0,
    exceptions=(Exception,),
    report=print,
    sleep=time.sleep,
):
    """Алдаа гарвал функцийг times удаа хүртэл дахин оролддог decorator."""
    if times < 1:
        raise ValueError(f"retry: times дор хаяж 1 байх ёстой, {times} ирлээ")

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as error:
                    last_error = error
                    report(
                        f"{func.__name__}: {attempt}/{times} "
                        f"оролдлого бүтэлгүйтэв: {error}"
                    )
                    if attempt < times:
                        sleep(delay)
            raise last_error

        return wrapper

    return decorator

Энэ бол Хичээл 5.3-ын Жишээ 2-ын retry, гурван нэмэлттэй. Нэгдүгээрт, times-ийн шалгалт retry-ийн гадна давхаргад, decorator тодорхойлогдохоос өмнө байна. Тэгэхээр @retry(times=0) мөр файл импортлогдох мөчид, тохиргооны цагт алдаа өгнө, функц хэзээ ч дуудагдахгүй байсан ч. Хоёрдугаарт, report. Гуравдугаарт, sleep: wrapper time.sleep-ийг шууд биш, үүргэвчин доторх sleep-ийг дуудна. Хэвийн хэрэглээнд энэ нь ижил time.sleep. Тест өөр функц өгөх боломжтой.

sleep-ийн анхдагч утга time.sleep нь def мөр ажиллах үед, модуль импортлогдох мөчид нэг удаа бодогддог. Энэ нь ердийн ба аюулгүй, учир нь time.sleep функц өөрчлөгддөггүй объект. Анхдагч утга list эсвэл dict байх үеийн урхийг Хичээл 7.4-т dataclass-тай хамт судална.

Шалгах цэг 3.

python

>>> from measurekit import retry
>>> @retry(times=0)
... def nothing():
...     pass
...
Traceback (most recent call last):
  ...
ValueError: retry: times дор хаяж 1 байх ёстой, 0 ирлээ

Алдаа функцийг тодорхойлох үед гарсан, бид nothing-ийг дуудаагүй байхад. Энэ бол тохиргооны цагийн шалгалт.

Алхам 4: гурвууланг хамт

demo.py файл үүсгэнэ:

python

from measurekit import retry, timed, timer


def make_flaky(fail_times):
    calls = 0

    def fetch():
        nonlocal calls
        calls += 1
        if calls <= fail_times:
            raise ConnectionError(f"{calls}-р дуудлага бүтэлгүйтэв")
        return "үнийн мэдээ ирлээ"

    return fetch


flaky_fetch = make_flaky(2)


@timed
def build_squares(n):
    return [x * x for x in range(n)]


@timed(label="Үнэ татах")
@retry(times=4, delay=0.2, exceptions=(ConnectionError,))
def fetch_prices():
    return flaky_fetch()


build_squares(1_000_000)

with timer("Нийлбэр") as t:
    total = sum(range(5_000_000))
print(f"Хадгалагдсан хэмжилт: {t['elapsed']:.4f} секунд")

print(fetch_prices())

make_flaky бол Хичээл 5.3-ын closure тоолуур. fetch_prices дээр хоёр decorator давхарласан: timed гадна, retry дотор. Хичээл 5.4-ийн дүрмээр timed хэрэглэгчийн мэдрэх бүх хугацааг хэмжинэ: бүтэлгүй оролдлогууд, тэдгээрийн хоорондох хүлээлтүүд, эцсийн амжилт. Ажиллуулна:

python demo.py

Миний компьютер дээрх нэг ажиллуулалт:

build_squares: 0.0421 секунд
Нийлбэр: 0.0729 секунд
Хадгалагдсан хэмжилт: 0.0729 секунд
fetch_prices: 1/4 оролдлого бүтэлгүйтэв: 1-р дуудлага бүтэлгүйтэв
fetch_prices: 2/4 оролдлого бүтэлгүйтэв: 2-р дуудлага бүтэлгүйтэв
Үнэ татах: 0.4013 секунд
үнийн мэдээ ирлээ

"Үнэ татах" 0.40 секунд орчим: хоёр бүтэлгүйтэл, тус бүрийн дараа 0.2 секунд хүлээлт. Хэрэв decorator-уудын дарааллыг солиход timed retry-ийн дотор очиж, оролдлого бүрийг тусад нь, хүлээлтгүйгээр хэмжинэ. Хүсвэл сольж үзээрэй. Retry-ийн мэдээнд “fetch_prices” гэж зөв нэр гарч байгааг анзаараарай: retry-ийн func нь анхны fetch_prices, timed-ийн wraps-аас өмнө.

Алхам 5: тестүүд

test_measurekit.py файл үүсгэнэ:

python

import pytest

from measurekit import retry, timed, timer


def silent(message):
    pass


def make_flaky(fail_times, error_type=ConnectionError):
    calls = 0

    def fetch():
        nonlocal calls
        calls += 1
        if calls <= fail_times:
            raise error_type(f"{calls}-р дуудлага бүтэлгүйтэв")
        return "өгөгдөл"

    return fetch, lambda: calls


def test_timed_keeps_identity():
    @timed(report=silent)
    def add(a, b):
        """Хоёр тоог нэмнэ."""
        return a + b

    assert add(2, 3) == 5
    assert add.__name__ == "add"
    assert add.__doc__ == "Хоёр тоог нэмнэ."
    assert add.__wrapped__(2, 3) == 5


def test_timed_without_parentheses_uses_function_name(capsys):
    @timed
    def square(x):
        return x * x

    assert square(4) == 16
    assert capsys.readouterr().out.startswith("square: ")


def test_timer_reports_error_and_reraises():
    messages = []
    with pytest.raises(ZeroDivisionError):
        with timer("Хуваах", report=messages.append):
            1 / 0
    assert len(messages) == 1
    assert "алдаатай дууслаа" in messages[0]


def test_timer_fills_result():
    with timer("Нийлбэр", report=silent) as result:
        sum(range(1000))
    assert result["ok"] is True
    assert result["elapsed"] >= 0


def test_retry_succeeds_after_failures():
    fetch, call_count = make_flaky(2)
    reliable = retry(times=3, report=silent)(fetch)
    assert reliable() == "өгөгдөл"
    assert call_count() == 3


def test_retry_gives_up_with_last_error():
    fetch, call_count = make_flaky(5)
    reliable = retry(times=3, report=silent)(fetch)
    with pytest.raises(ConnectionError, match="3-р дуудлага"):
        reliable()
    assert call_count() == 3


def test_retry_ignores_other_error_types():
    fetch, call_count = make_flaky(1, error_type=ValueError)
    reliable = retry(times=3, exceptions=(ConnectionError,), report=silent)(fetch)
    with pytest.raises(ValueError):
        reliable()
    assert call_count() == 1


def test_retry_waits_between_attempts():
    waits = []
    fetch, _ = make_flaky(5)
    reliable = retry(times=3, delay=0.5, report=silent, sleep=waits.append)(fetch)
    with pytest.raises(ConnectionError):
        reliable()
    assert waits == [0.5, 0.5]


def test_retry_rejects_zero_times():
    with pytest.raises(ValueError):
        retry(times=0)

Тестүүдийн дизайны гол санаануудыг товч тэмдэглэе. silent функц юу ч хийдэггүй report бөгөөд тестийн гаралтыг цэвэр байлгана. Тестийн make_flaky closure тоолуурын хажуугаар lambda: calls гэсэн хоёр дахь функц буцаадаг: тэр ижил үүргэвчнээс calls-ийг уншдаг тул тест функц хэдэн удаа дуудагдсаныг шалгаж чадна. Хичээл 4.2-ын Жишээ 3-ын "нэг хайрцгийг хуваалцах хэд хэдэн функц" энд хэрэгжиж байна. Хоёр дахь тест pytest-ийн capsys fixture-ийг ашиглаж, анхдагч print-ийн гаралтыг барьж шалгана. Сүүлийн өмнөх тест sleep=waits.append өгсөн: retry хүлээхийн оронд хүлээх хугацааг list-д бичнэ. Тэгэхээр 0.5 секундийн хоёр хүлээлттэй тест агшин зуур дуусна. Хугацааны тодорхой утгыг шалгадаг тест байхгүйг анзаараарай: компьютер бүрт хугацаа өөр, тиймээс ийм тест заримдаа унах тогтворгүй тест болно. Бид зөвхөн бүтэц, шошго, дугаарлалт, тоолол, хүлээлтийн хүсэлтийг шалгасан.

Ажиллуулна:

pytest -q

.........                                                            [100%]
9 passed in 0.03s

Алхам 6: commit

git status
git add measurekit.py test_measurekit.py demo.py
git commit -m "Add measurekit: timer, timed and retry with tests"

Энэ төсөлд үүсгэгдсэн том файл байхгүй тул .gitignore шаардлагагүй. Гэвч pytest .pytest_cache хавтас, Python __pycache__ хавтас үүсгэсэн байж магадгүй. Course 4-ийн repository-д .gitignore эдгээрийг аль хэдийн хамарсан байх ёстой. git status-д харагдвал repository-ийн .gitignore-д нэмнэ. Branch-ийг main руу merge хийх эсэх нь таны сонголт.

Бүрэн код: measurekit.py

python

"""Хугацаа хэмжих ба дахин оролдох жижиг хэрэгслүүд."""

import time
from contextlib import contextmanager
from functools import partial, wraps


@contextmanager
def timer(label="Блок", report=print):
    """with блокийн ажиллах хугацааг хэмжиж, report-оор мэдээлнэ."""
    result = {"label": label, "elapsed": None, "ok": False}
    start = time.perf_counter()
    try:
        yield result
        result["ok"] = True
    finally:
        result["elapsed"] = time.perf_counter() - start
        status = "" if result["ok"] else " (алдаатай дууслаа)"
        report(f"{label}: {result['elapsed']:.4f} секунд{status}")


def timed(func=None, *, label=None, report=print):
    """Функцийн дуудлага бүрийн хугацааг хэмждэг decorator."""
    if func is None:
        return partial(timed, label=label, report=report)
    if not callable(func):
        raise TypeError("timed: шошгыг label=... гэж түлхүүр үгээр өгнө үү")

    name = label or func.__name__

    @wraps(func)
    def wrapper(*args, **kwargs):
        with timer(name, report=report):
            return func(*args, **kwargs)

    return wrapper


def retry(
    times=3,
    delay=0.0,
    exceptions=(Exception,),
    report=print,
    sleep=time.sleep,
):
    """Алдаа гарвал функцийг times удаа хүртэл дахин оролддог decorator."""
    if times < 1:
        raise ValueError(f"retry: times дор хаяж 1 байх ёстой, {times} ирлээ")

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as error:
                    last_error = error
                    report(
                        f"{func.__name__}: {attempt}/{times} "
                        f"оролдлого бүтэлгүйтэв: {error}"
                    )
                    if attempt < times:
                        sleep(delay)
            raise last_error

        return wrapper

    return decorator

demo.py ба test_measurekit.py нь Алхам 4 ба 5-д бүрэн хэлбэрээрээ байгаа.

Сонголтот өргөтгөлүүд

Хүсвэл дараах өргөтгөлүүдээс сонгож туршиж болно. Тэдгээр нь заавал биш бөгөөд дараагийн хичээлүүд тэдгээрийг хийсэн гэж огт үзэхгүй.

Дуудлагын статистик. timed-д stats=True тохиргоо нэмж, Хичээл 4.3-ын make_tracker хуримтлуулагчийг wrapper-ийн гадна, timed-ийн гэрт байрлуулбал функц бүр өөрийн дуудлагын тоо, дундаж, хамгийн их хугацааг хуримтлуулна. Статистикийг wrapper-т attribute болгон, жишээлбэл wrapper.stats = tracker, хавсаргаж хэрэглэгчид өгч болно. Функц бол объект тул түүнд attribute нэмж болдог.

Exponential backoff. retry-д backoff=1.0 тохиргоо нэмж, оролдлого бүрийн дараа delay-г түүгээр үржүүлбэл 0.2, 0.4, 0.8 гэх мэт өсөн нэмэгдэх хүлээлт гарна. Шинэ тест бичихэд sleep=waits.append загвар бэлэн: хүлээлтийн жагсаалт [0.2, 0.4]-тэй тэнцүү эсэхийг шалгана.

Давтан хэрэглэгдэх Timer. timer-ийг Хичээл 6.2-ын загвараар class болгон дахин бичиж, нэг объектыг loop дотор давтан хэрэглэх, бүх хэмжилтийн жагсаалтыг хадгалах, reset() method нэмэх боломжтой болгоно. Generator хувилбартай харьцуулж, аль нь ямар тохиолдолд илүү гэдгийг өөрийн туршлагаар дүгнэнэ.

Түгээмэл алдаанууд

TypeError: timed: шошгыг label=... гэж түлхүүр үгээр өгнө үү. Энэ бол бидний өөрсдийн бичсэн алдааны мессеж. @timed("Тайлан") гэж шошгыг байрлалаар өгсөн үед гарна. Энэ хамгаалалтгүй байсан бол Python "Тайлан" string-ийг боож, дараа нь wrapper-ийг decorator болгон хэрэглэх үед string-ийг дуудах гэж оролдож “TypeError: ‘str’ object is not callable” гэсэн ойлгомжгүй алдаа өгөх байсан. Сайн сан хэрэглэгчийн түгээмэл алдааг урьдчилан таамаглаж, юу буруу болсныг, хэрхэн засахыг тодорхой хэлдэг. Шийдэл: @timed(label="Тайлан").

ModuleNotFoundError: No module named ‘measurekit’. “measurekit нэртэй модуль олдсонгүй” гэсэн утгатай. pytest эсвэл demo.py-г measure-tools хавтаснаас өөр газраас ажиллуулсан, эсвэл файлын нэрийг өөрөөр бичсэн үед гарна. Хичээл 3.4-ийн адил шийдэл: terminal-ийн одоогийн хавтсыг шалгах. Тест файл ба сан нэг хавтаст байгаа тул pytest тест файлын хавтсыг модуль хайх замд нэмж, импорт ажиллана.

Тест заримдаа унах. Хэрэв та тестэд assert result["elapsed"] < 0.01 гэх мэт хугацааны хязгаар шалгавал тест таны компьютер дээр ихэнхдээ давах ч, компьютер ачаалалтай үед эсвэл өөр машин дээр унана. Ийм тогтворгүй тест хамгийн хортой: багийнхан тэдгээрт итгэхээ больдог. Шийдэл нь хугацааны тодорхой утгыг огт шалгахгүй байх: бүтэц, тоолол, дарааллыг шалгах. Хугацааг шалгах шаардлагатай бол sleep шиг "цаг"-ийг ч тохиргоогоор өгч, тестэд хуурамч цаг ашиглана.

Тест удаан болох. Retry-ийн тестэд sleep тохиргоо өгөхөө мартвал delay=0.5-тай тест жинхэнээр хүлээнэ. Нэг тест секунд, тестийн багц минут болно. pytest-ийн гаралтын төгсгөлд “9 passed in 1.03s” гэх мэт нийт хугацаа гарах тул түүнийг ажиглаарай. Бүх тест 0.1 секундээс хурдан дуусах ёстой. Хүлээлттэй код бүрийг тестлэхэд sleep шиг хамаарлыг тохиргоогоор өгөх загвар хэрэгтэй.

ValueError: retry: times дор хаяж 1 байх ёстой, 0 ирлээ. Энэ ч бидний бичсэн мессеж бөгөөд зориудаар файл импортлогдох үед гарна. Хэрэв таны програм ямар ч функц дуудагдахаас өмнө энэ алдаагаар унавал @retry мөрүүдийн times утгыг шалгаарай. times тохиргооны файл эсвэл орчны хувьсагчаас уншигддаг бол тэр утга тоо болж хөрвүүлэгдсэн эсэхийг ч шалгаарай.

Товч дүгнэлт

Бид гурван хэрэгсэлтэй жижиг сан бүтээлээ. timer бол try/finally-тай @contextmanager бөгөөд yield-ийн дараах “ok” тэмдэглэлээр блок амжилттай байсан эсэхийг ялгаж, үр дүнг dict-ээр хэрэглэгчид өгдөг. timed бол @timed ба @timed(label=...) хоёр хэлбэрийг дэмждэг decorator бөгөөд func ирээгүй үед partial-аар өөрийгөө тохируулж буцаадаг, хэмжилтийг timer-ийг дахин ашиглан хийдэг, байрлалаар өгсөн шошгыг тодорхой мессежээр барьдаг. retry бол тохиргооны цагт times-ийг шалгадаг, алдааны төрлийг шүүдэг, оролдлогын хооронд хүлээдэг гурван давхар decorator. Гурвуулаа wraps ашигладаг, гурвуулаа мэдээллээ report функцээр гаргадаг, retry хүлээлтээ sleep функцээр хийдэг. Эдгээр нь хэрэгслийн гадаад хамаарлыг тохиргоогоор өгөх дизайн бөгөөд есөн тест агшин зуур, хуурамч report ба sleep-ээр шалгагдсан. Давхарласан decorator-уудын дараалал юуг хэмжихийг шийдсэн. Ажил branch дээр commit хийгдсэн.

Энэ хичээлээр Түвшин 2 дуусч байна. Функц таны хувьд нууцгүй боллоо: тэр утга, тэр үүргэвчтэй, тэр боогдож, тохируулагдаж, давхарлагдаж чадна. Дараагийн түвшин OOP-ийн эзэн. Бид class-уудын гүн рүү орж, тэдгээрийг Python-ий бэлэн төрлүүд шиг байгалийн мэт ажилладаг болгоно. Эхний хичээл classmethod: class-ийн өөр төрлийн method бөгөөд хамгийн чухал хэрэглээ нь "өөр хэлбэрийн өгөгдлөөс объект үүсгэх" буюу alternative constructor. Энэ бүтээлийн timed func is None шалгалтаар хоёр өөр замаар объект үүсгэдэг шиг, classmethod class-д олон хаалга нээнэ: string-ээс, dict-ээс, файлаас.