يتم وصف Apis بالفئات ، على سبيل المثال
class Categories(JsonEndpoint):
url = "http://127.0.0.1:8888/categories"
params = {"page": range(100), "language": "en"}
headers = {"User-Agent": get_user_agent}
results_key = "*.slug"
categories = Categories()
class Posts(JsonEndpoint):
url = "http://127.0.0.1:8888/categories/{category}/posts"
params = {"page": range(100), "language": "en"}
url_params = {"category": categories.iter_results()}
results_key = "posts"
async def comments(self, post):
comments = Comments(
self.session,
url_params={"category": post.url.params["category"], "id": post["id"]},
)
return [comment async for comment in comments]
posts = Posts()
يمكن أن تحتوي المعلمات و url_params على وظائف (كما هو الحال هنا get_user_agent - تُرجع وكيل مستخدم عشوائي) ، والمدى ، والمكررات ، والمكررات المنتظرة وغير المتزامنة (حتى تتمكن من ربطهم معًا).
يمكن أن تحتوي رؤوس المعلمات وملفات تعريف الارتباط أيضًا على وظائف ويمكن انتظارها.
تعرض فئة api في المثال أعلاه مصفوفة من الكائنات التي تحتوي على سبيكة ، وسيعيد المكرر هذه العناصر بالضبط. من خلال تمرير هذا المكرر في url_params للمنشورات ، سوف يمر المكرر بشكل متكرر عبر جميع الفئات وجميع الصفحات في كل منها. سيتم إحباطه عندما يواجه خطأ 404 أو خطأ آخر وينتقل إلى الفئة التالية.
وتحتوي المستودعات على مثال لخادم aiohttp لهذه الفئات بحيث يمكن اختبار كل شيء.
بالإضافة إلى الحصول على المعلمات ، يمكنك تمريرها كبيانات أو json وتعيين طريقة أخرى.
results_key منقط وسيحاول سحب المفاتيح من النتائج. على سبيل المثال "التعليقات. *. النص" سيعيد نص كل تعليق من المصفوفة الموجودة داخل التعليقات.
يتم تغليف النتائج في غلاف يحتوي على خصائص url والمعلمات. عنوان url مشتق من سلسلة تحتوي أيضًا على معلمات. وبالتالي ، يمكنك معرفة المعلمات التي تم استخدامها للحصول على هذه النتيجة ، وهذا موضح في طريقة التعليقات.
هناك أيضًا فئة حوض أساسية للتعامل مع النتائج. على سبيل المثال ، طيها في mq أو قاعدة بيانات. يعمل في مهام منفصلة ويتلقى البيانات عبر asyncio.Queue.
class LoggingSink(Sink):
def transform(self, obj):
return repr(obj)
async def init(self):
from loguru import logger
self.logger = logger
async def process(self, obj):
self.logger.info(obj)
return True
sink = LoggingSink(num_tasks=1)
مثال على أبسط بالوعة. تسمح لنا طريقة التحويل بإجراء بعض التلاعبات مع الكائن وإرجاع لاشيء إذا لم يناسبنا ذلك. أولئك. في السمات ، يمكنك أيضًا إجراء التحقق من الصحة.
Sink هو مدير سياق غير متزامن ، والذي ، عند الخروج منه ، سينتظر نظريًا حتى تتم معالجة جميع الكائنات في قائمة الانتظار ، ثم يلغي مهامه.
وأخيرًا ، لربط كل ذلك معًا ، قمت بإنشاء فصل دراسي للعمال. يقبل نقطة نهاية واحدة والعديد من الأحواض. على سبيل المثال،
worker = Worker(endpoint=posts, sinks=[loggingsink, mongosink])
worker.run()
سيعمل التشغيل على تشغيل asyncio.run_until_complete لخط أنابيب العامل. كما أن لديها طريقة تحويل.
هناك أيضًا فئة WorkerGroup التي تتيح لك إنشاء العديد من العمال في وقت واحد وإنشاء لقاء غير متزامن لهم.
يحتوي الكود على مثال لخادم يقوم بتوليد البيانات من خلال الخادع والمعالجات لنقاط النهاية الخاصة به. أعتقد أن هذا هو الأكثر وضوحا.
كل هذا في مرحلة مبكرة من التطوير وحتى الآن غالبًا ما قمت بتغيير واجهة برمجة التطبيقات. ولكن يبدو الآن أنه قد وصل إلى الشكل الذي ينبغي أن يبدو عليه. سأقوم فقط بدمج الطلبات والتعليقات في الكود الخاص بي.