بناء مُزخرف retry

بناء مُزخرف retry — Build a Retry Decorator

في التطوير الحقيقي، العمليات تفشل: طلبات الشبكة تنتهي مهلتها، قواعد البيانات تكون مشغولة، الخدمات الخارجية تتعطل. الحل الاحترافي هو الإعادة التلقائية عند الفشل (Retry) — وأفضل مكان لهذا المنطق هو مُزخرف قابل للإعادة الاستخدام.

ستبني هذا المُزخرف خطوة بخطوة، بدءاً من النسخة الأبسط ووصولاً لنسخة إنتاجية متكاملة. كل خطوة تبني على السابقة.

الخطوة 1 — مُزخرف بسيط (بدون إعداد)

أولاً، نبني مُزخرفاً يُعيد المحاولة مرتين إضافيتين عند أي فشل — من غير معاملات (لا factory بعد).

main.go

import functools


def retry(fn):
    """يُعيد المحاولة حتى 3 مرات عند أي استثناء — Retries up to 3 times on any exception."""
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        last_error = None
        for attempt in range(3):  # 0، 1، 2 — ثلاث محاولات
            try:
                return fn(*args, **kwargs)   # حاوِل — try
            except Exception as e:
                last_error = e
                print(f"[RETRY] محاولة {attempt + 1}/3 فشلت: {e}")
        raise last_error  # استنفدنا المحاولات — exhausted retries

    return wrapper


# محاكاة دالة تفشل في البداية — Simulating a function that fails at first
call_count = 0

@retry
def flaky_function():
    global call_count
    call_count += 1
    if call_count < 3:
        raise ValueError(f"فشل مؤقت #{call_count} — Temporary failure")
    return f"نجحت في المحاولة {call_count}"


result = flaky_function()
print(f"النتيجة: {result}")

Output:

المُزخرف يلتقط Exception، يطبع رسالة، ويُعيد المحاولة. إذا استنفد كل المحاولات يُعيد إطلاق آخر خطأ لكي يُعالجه المستدعي.

الخطوة 2 — decorator factory مع `times`

المُزخرف الثابت غير مرن. نريد @retry(times=5). هذا يتطلب decorator factory — دالة تُنتج مُزخرفاً.

تحدي — Challenge

import functools


def retry(times=3):
    """decorator factory — يُرجع مُزخرفاً بعدد محاولات مخصص — Returns decorator with custom retry count."""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # أكمل: أضف منطق الإعادة مع range(times) — Complete: add retry logic with range(times)
            pass
        return wrapper
    return decorator


call_count = 0

@retry(times=4)
def unstable():
    global call_count
    call_count += 1
    if call_count < 4:
        raise RuntimeError(f"خطأ {call_count}")
    return f"نجحت في المحاولة {call_count}"


print(unstable())

الخطوة 3 — الإمساك بالاستثناء والإعادة الصحيحة

بنينا الهيكل. الآن نتأكد أن المُزخرف يُعيد رمي الاستثناء إذا استنفدت كل المحاولات، ولا يبتلع الخطأ صامتاً.

main.go

import functools


def retry(times=3):
    """مُزخرف retry كامل — Full retry decorator."""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    last_exc = e
                    print(f"[RETRY] {fn.__name__} محاولة {attempt}/{times}: {e}")
            # استنفدنا المحاولات — exhausted all attempts
            print(f"[RETRY] {fn.__name__} فشلت بعد {times} محاولات — failed after {times} attempts")
            raise last_exc
        return wrapper
    return decorator


# اختبار: دالة تفشل دائماً — Test: always-failing function
@retry(times=3)
def always_fails():
    raise ConnectionError("تعذّر الاتصال — Connection refused")


# اختبار: الإعادة ثم النجاح — Test: retry then succeed
attempts = 0

@retry(times=4)
def eventually_works():
    global attempts
    attempts += 1
    if attempts < 3:
        raise TimeoutError(f"انتهت المهلة #{attempts}")
    return "وُصل بنجاح — Connected successfully"


try:
    always_fails()
except ConnectionError as e:
    print(f"استُلم الخطأ: {e}")

print()
attempts = 0
print(eventually_works())

Output:

الخطوة 4 — إضافة Backoff أسّي

إعادة المحاولة فوراً قد تُرهق الخادم. في الإنتاج نضيف تأخير يتضاعف بعد كل فشل: 1s، 2s، 4s…

تحدي — Challenge

import functools
import time


def retry(times=3, delay=1):
    """retry مع backoff أسّي — retry with exponential backoff."""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            last_exc = None
            current_delay = delay  # الانتظار الأولي — initial wait

            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    last_exc = e
                    if attempt < times:
                        # أكمل: اطبع رسالة الانتظار ثم نم — Complete: print wait message then sleep
                        pass
                    # أكمل: ضاعف current_delay — Complete: double current_delay

            raise last_exc
        return wrapper
    return decorator


call_n = 0

@retry(times=3, delay=1)
def needs_backoff():
    global call_n
    call_n += 1
    if call_n < 3:
        raise IOError("خطأ")
    return "نجحت!"


# ملاحظة: في الاختبار نحتاج time.sleep حقيقي هنا
# Note: real time.sleep needed here
result = needs_backoff()
print(result)

الخطوة 5 — تصفية الاستثناءات

أحياناً تريد الإعادة فقط عند استثناءات معينة. خطأ ValueError (بيانات خاطئة) لا يستحق إعادة — أما ConnectionError فيستحق.

تحدي — Challenge

import functools


def retry(times=3, delay=0, exceptions=(Exception,)):
    """retry مع تصفية الاستثناءات — retry with exception filtering."""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    # أكمل: إذا لم يكن e من أنواع exceptions، أعِد إطلاقه فوراً
                    # Complete: if e is not in exceptions types, re-raise immediately
                    last_exc = e
                    if attempt < times:
                        print(f"[RETRY] {attempt}/{times}: يعيد — retrying")
            raise last_exc
        return wrapper
    return decorator


retries = 0

# يُعيد فقط على ConnectionError — Only retries on ConnectionError
@retry(times=2, delay=0, exceptions=(ConnectionError,))
def fetch():
    global retries
    retries += 1
    if retries < 2:
        raise ConnectionError("فشل الاتصال")
    return "نجحت"


# لا يُعيد على ValueError — Does NOT retry on ValueError
@retry(times=3, delay=0, exceptions=(ConnectionError,))
def validate(data):
    if not data:
        raise ValueError("بيانات خاطئة — Bad data")
    return "صحيح"


print(fetch())
print("---")
try:
    validate("")
except ValueError as e:
    print(f"خطأ فوري: {e}")

النسخة الكاملة — Full Production Version

الآن نجمع كل الخطوات في مُزخرف واحد إنتاجي:

main.go

import functools
import time
from typing import Tuple, Type


def retry(
    times: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
    """
    مُزخرف retry إنتاجي — Production-grade retry decorator.

    :param times: عدد المحاولات — number of attempts
    :param delay: الانتظار الأولي بالثواني — initial delay in seconds
    :param backoff: معامل تضاعف التأخير — delay multiplier (1.0 = no backoff)
    :param exceptions: أنواع الاستثناءات المسموح بإعادتها — exception types to retry on
    """
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            current_delay = delay
            last_exc: Exception = RuntimeError("لم تُجرَ أي محاولة — no attempt made")

            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)

                except exceptions as e:
                    last_exc = e
                    if attempt < times:
                        print(
                            f"[RETRY] {fn.__name__} "
                            f"محاولة {attempt}/{times} — فشلت: {e} "
                            f"(انتظر {current_delay:.1f}s)"
                        )
                        time.sleep(current_delay)
                        current_delay *= backoff  # تضاعف التأخير — double the delay

                except Exception:
                    raise  # استثناء خارج القائمة — outside exceptions list

            print(f"[RETRY] {fn.__name__} استنفدت {times} محاولات — exhausted {times} attempts")
            raise last_exc

        return wrapper
    return decorator


# اختبار النسخة الكاملة — Test full version
attempt_n = 0

@retry(times=4, delay=0, backoff=1.0, exceptions=(ConnectionError, TimeoutError))
def connect_to_api(url: str) -> str:
    global attempt_n
    attempt_n += 1
    if attempt_n < 3:
        raise ConnectionError(f"تعذّر الاتصال بـ {url}")
    return f"متصل بـ {url}"


attempt_n = 0
result = connect_to_api("https://api.example.com")
print(f"\nالنتيجة: {result}")

# اختبار الاستثناء الفوري — Test immediate exception
@retry(times=3, delay=0, exceptions=(ConnectionError,))
def parse_data(raw: str) -> dict:
    if not raw.strip():
        raise ValueError("بيانات فارغة — empty data")  # لن يُعاد
    return {"data": raw}


try:
    parse_data("")
except ValueError as e:
    print(f"خطأ فوري (صحيح): {e}")

Output:

الخلاصة — Summary

المُزخرف retry الذي بنيته يجسّد كل ما تعلمته في هذا الفصل:

الدوال كقيم — decorator تأخذ fn وتُرجع wrapper
decorator factory — retry(times, delay, ...) تُنتج مُزخرفاً قابلاً للإعداد
functools.wraps — يحفظ اسم ووثائق الدالة الأصلية
*args, **kwargs — يجعل wrapper مرناً مع أي توقيع
Type hints — توثيق واضح لكل معامل ونوعه
معالجة الاستثناءات — التمييز بين ما يستحق الإعادة وما لا يستحق

هذا النمط موجود في مكتبات Python الشهيرة: tenacity، backoff، retry، وurllib3. الآن تفهم كيف تبنيه من الصفر — وهذا أهم من حفظ API مكتبة.

بناء مُزخرف retry