دليل برمجي لبناء مجموعة أدوات تطوير برمجيات (SDK) متزامنة عالية الأداء بلغة بايثون مع تحديد معدل الاستجابة، والتخزين المؤقت في الذاكرة، والمصادقة

يقدم هذا الدليل الشامل خطوات بناء مجموعة أدوات تطوير برمجيات (SDK) متزامنة بلغة بايثون، مُصممة لتلبية متطلبات الإنتاج، مع التركيز على الأداء والكفاءة. سنتناول في هذا الدليل مكونات أساسية تضمن استقرار وفعالية مجموعة الأدوات، بدءًا من تثبيت وتكوين مكتبات HTTP المتزامنة الأساسية ووصولاً إلى إدارة الاستثناءات وإمكانية التوسعة.

1. التثبيت والتكوين

نقوم أولاً بتثبيت وتكوين بيئة العمل اللازمة باستخدام المكتبات التالية:

• asyncio: لإدارة العمليات المتزامنة.
• aiohttp: لإجراء طلبات HTTP المتزامنة.
• nest_asyncio: لتمكين تشغيل حلقة الأحداث داخل بيئات مثل Jupyter Notebook أو Google Colab.
• dataclasses: لإنشاء نماذج بيانات منظمة.
• hashlib: لإنشاء مفاتيح التجزئة للتخزين المؤقت.
• datetime: لإدارة تواريخ وأوقات انتهاء صلاحية البيانات المخزنة مؤقتًا.
• logging: لإدارة السجلات.

يمكن تثبيت هذه المكتبات باستخدام أمر pip:

pip install aiohttp nest_asyncio

2. تصميم كائن الاستجابة

سنبدأ بتعريف كائن APIResponse لاستقبال الاستجابات من الخادم، وهو كائن بيانات منظم (dataclass) يحتوي على:

• data: بيانات الاستجابة.
• status_code: رمز حالة الاستجابة (مثلاً: 200 للنجاح).
• headers: عناوين الاستجابة.
• timestamp: وقت استلام الاستجابة.

from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Any, Dict

@dataclass
class APIResponse:
    """كائن بيانات منظم لاستجابة API"""
    data: Any
    status_code: int
    headers: Dict[str, str]
    timestamp: datetime

    def to_dict(self) -> Dict:
        return asdict(self)

3. محدد معدل الاستجابة (Rate Limiter)

لتجنب تجاوز حدود معدل الاستجابة للخادم، سنستخدم RateLimiter الذي يطبق سياسة “Token Bucket” لتحديد عدد الطلبات المسموح بها خلال فترة زمنية معينة:

import time

class RateLimiter:
    """محدد معدل الاستجابة باستخدام سياسة Token Bucket"""
    def __init__(self, max_calls: int = 100, time_window: int = 60):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = []

    def can_proceed(self) -> bool:
        now = time.time()
        self.calls = [call_time for call_time in self.calls if now - call_time < self.time_window]
        if len(self.calls) < self.max_calls:
            self.calls.append(now)
            return True
        return False

    def wait_time(self) -> float:
        if not self.calls:
            return 0
        return max(0, self.time_window - (time.time() - self.calls[0]))

4. التخزين المؤقت (Cache)

لتحسين الأداء، سنستخدم Cache لتخزين الاستجابات مؤقتًا في الذاكرة مع تحديد وقت انتهاء صلاحيتها (TTL):

import hashlib
from datetime import timedelta

class Cache:
    """تخزين مؤقت بسيط في الذاكرة مع TTL"""
    def __init__(self, default_ttl: int = 300):
        self.cache = {}
        self.default_ttl = default_ttl

    def _generate_key(self, method: str, url: str, params: Dict = None) -> str:
        key_data = f"{method}:{url}:{json.dumps(params or {}, sort_keys=True)}"
        return hashlib.md5(key_data.encode()).hexdigest()

    def get(self, method: str, url: str, params: Dict = None) -> Optional[APIResponse]:
        key = self._generate_key(method, url, params)
        if key in self.cache:
            response, expiry = self.cache[key]
            if datetime.now() < expiry:
                return response
            del self.cache[key]
        return None

    def set(self, method: str, url: str, response: APIResponse, params: Dict = None, ttl: int = None):
        key = self._generate_key(method, url, params)
        expiry = datetime.now() + timedelta(seconds=ttl or self.default_ttl)
        self.cache[key] = (response, expiry)

5. مجموعة أدوات تطوير البرمجيات (AdvancedSDK)

تجمع هذه الفئة جميع المكونات السابقة معًا:

import aiohttp
import logging
from typing import Dict, Optional

class AdvancedSDK:
    """مجموعة أدوات تطوير برمجيات متقدمة مع أنماط بايثون الحديثة"""
    def __init__(self, base_url: str, api_key: str = None, rate_limit: int = 100):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session = None
        self.rate_limiter = RateLimiter(max_calls=rate_limit)
        self.cache = Cache()
        self.logger = self._setup_logger()

    def _setup_logger(self) -> logging.Logger:
        # ... ( نفس  الكود من المثال الأصلي ) ...

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    def _get_headers(self) -> Dict[str, str]:
        # ... ( نفس  الكود من المثال الأصلي ) ...

    async def _make_request(self, method: str, endpoint: str, params: Dict = None, data: Dict = None, use_cache: bool = True) -> APIResponse:
        # ... ( نفس  الكود من المثال الأصلي  مع تعديلات طفيفة في التعليقات ) ...

    async def get(self, endpoint: str, params: Dict = None, use_cache: bool = True) -> APIResponse:
        return await self._make_request('GET', endpoint, params=params, use_cache=use_cache)

    async def post(self, endpoint: str, data: Dict = None) -> APIResponse:
        return await self._make_request('POST', endpoint, data=data, use_cache=False)

    async def put(self, endpoint: str, data: Dict = None) -> APIResponse:
        return await self._make_request('PUT', endpoint, data=data, use_cache=False)

    async def delete(self, endpoint: str) -> APIResponse:
        return await self._make_request('DELETE', endpoint, use_cache=False)

6. مثال عملي (Demo)

يوضح هذا المثال كيفية استخدام AdvancedSDK:

import asyncio

async def demo_sdk():
    # ... ( نفس الكود من المثال الأصلي  مع تعديلات طفيفة في التعليقات ) ...

async def run_demo():
    await demo_sdk()

# ... ( نفس الكود من المثال الأصلي  لـ  nest_asyncio ) ...

7. نمط Builder

يُسهّل هذا النمط تكوين AdvancedSDK:

class SDKBuilder:
    """نمط Builder لتكوين SDK"""
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.config = {}

    def with_auth(self, api_key: str):
        self.config['api_key'] = api_key
        return self

    def with_rate_limit(self, calls_per_minute: int):
        self.config['rate_limit'] = calls_per_minute
        return self

    def build(self) -> AdvancedSDK:
        return AdvancedSDK(self.base_url, **self.config)

الخاتمة

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

المصدر: MarkTechPost