ما يجب أن يكون عدم التزامن في بايثون

في السنوات القليلة الماضية، و asyncالبرمجة غير المتزامنة الكلمة ودلالات وتخلل العديد من لغات البرمجة شعبية: جافا سكريبت ، الصدأ ، C # ، و غيرها الكثير . بالطبع ، Python تمتلكها أيضًا async/await، تم تقديمها في Python 3.5.



في هذه المقالة ، أرغب في مناقشة مشكلات الشفرة غير المتزامنة ، والتكهن بالبدائل ، واقتراح نهج جديد لدعم التطبيقات المتزامنة وغير المتزامنة في نفس الوقت.

لون الوظيفة

عندما يتم تضمين الوظائف غير المتزامنة في لغة البرمجة ، فإنها تنقسم أساسًا إلى قسمين. تظهر الوظائف الحمراء (أو غير متزامنة) وتظل بعض الوظائف زرقاء (متزامنة).



تكمن المشكلة الرئيسية في أن الوظائف الزرقاء لا يمكنها استدعاء الوظائف الحمراء ، ولكن من المحتمل أن تؤدي الوظائف الحمراء إلى وظائف زرقاء. في Python ، على سبيل المثال ، هذا صحيح جزئيًا: يمكن للوظائف غير المتزامنة فقط استدعاء الوظائف المتزامنة غير المحظورة. لكن من المستحيل تحديد ما إذا كانت الوظيفة محظورة أم لا من الوصف. بايثون هي لغة برمجة.



يؤدي هذا الانقسام إلى تقسيم اللغة إلى مجموعتين فرعيتين: متزامنة وغير متزامنة. تم إصدار Python 3.5 منذ أكثر من خمس سنوات ، لكنه asyncلا يزال غير مدعوم جيدًا مثل قدرات Python المتزامنة.



يمكنك قراءة المزيد عن الألوان الوظيفية في هذه المقالة الرائعة .

كود مكرر

الألوان المختلفة للوظائف تعني تكرار الكود في الممارسة.



تخيل أنك تقوم بتطوير أداة CLI لاسترداد حجم صفحة الويب وتريد الحفاظ على كل من الطرق المتزامنة وغير المتزامنة للقيام بذلك. على سبيل المثال ، هذا ضروري إذا كنت تكتب مكتبة ولا تعرف كيف سيتم استخدام التعليمات البرمجية الخاصة بك. ولا يتعلق الأمر بمكتبات PyPI فحسب ، بل يتعلق أيضًا بمكتباتنا الخاصة ذات المنطق المشترك لمختلف الخدمات ، المكتوبة ، على سبيل المثال ، في Django و aiohttp. على الرغم من أنه ، بالطبع ، تتم كتابة التطبيقات المستقلة في الغالب إما بشكل متزامن فقط أو بشكل غير متزامن فقط.



لنبدأ بالرمز الكاذب المتزامن:

def fetch_resource_size(url: str) -> int:
    response = client_get(url)
    return len(response.content)

يبدو جميل. الآن دعونا نلقي نظرة على التناظرية غير المتزامنة:

async def fetch_resource_size(url: str) -> int:
    response = await client_get(url)
    return len(response.content)

بشكل عام ، هذا هو نفس الكود ، ولكن مع إضافة الكلمات asyncو await. ولم أقم باختلاقه - قارن أمثلة التعليمات البرمجية في البرنامج التعليمي على httpx:

• كود متزامن .
• كود غير متزامن .

هناك نفس الصورة بالضبط.

التجريد والتكوين

وتبين أن تحتاج إلى إعادة كتابة كافة التعليمات البرمجية متزامن وترتيب هنا وهناك asyncو awaitحتى يتسنى للبرنامج يصبح غير متزامن.



يمكن أن يساعد مبدأان في حل هذه المشكلة. أولاً ، دعنا نعيد كتابة الشفرة الزائفة الحتمية إلى وظيفية. سيسمح لك ذلك برؤية الصورة بشكل أكثر وضوحًا.

def fetch_resource_size(url: str) -> Abstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

أنت تسأل ما هي هذه الطريقة .map، وماذا تفعل. هذه هي الطريقة التي يحدث بها تكوين التجريدات المعقدة والوظائف الصافية بأسلوب وظيفي. يتيح لك هذا إنشاء تجريد جديد بحالة جديدة من حالة موجودة. لنفترض أنها client_get(url)عادت في البداية Abstraction[Response]، .map(lambda response: len(response.content))وتحول المكالمة الاستجابة إلى المثيل المطلوب Abstraction[int].



يصبح من الواضح ما يجب القيام به بعد ذلك. لاحظ كيف انتقلنا بسهولة من بضع خطوات مستقلة إلى استدعاءات دالة متسلسلة. بالإضافة إلى ذلك ، قمنا بتغيير نوع الاستجابة: الآن تقوم الوظيفة بإرجاع بعض التجريد.



دعنا نعيد كتابة الكود للعمل مع الإصدار غير المتزامن:

def fetch_resource_size(url: str) -> AsyncAbstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

الشيء الوحيد المختلف هو نوع الإرجاع - AsyncAbstraction. باقي الكود هو نفسه تمامًا. لم تعد بحاجة إلى استخدام الكلمات الرئيسية asyncو await. awaitلا يستخدم على الإطلاق ( من أجل هذا بدأ كل شيء ) ، وبدونه لا جدوى من ذلك async.



آخر شيء هو تحديد العميل الذي نحتاجه: غير متزامن أو متزامن.

def fetch_resource_size(
    client_get: Callable[[str], AbstactionType[Response]],
    url: str,
) -> AbstactionType[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

client_getهي الآن وسيطة من النوع القابل للاستدعاء تأخذ سلسلة URL كمدخلات وتعيد نوعًا ما AbstractionTypeفوق الكائن Response. AbstractionType- إما Abstractionأو AsyncAbstractionمن الأمثلة السابقة.



عندما نمرر Abstraction، يعمل الرمز بشكل متزامن ، عندما AsyncAbstraction- يبدأ نفس الرمز تلقائيًا في العمل بشكل غير متزامن.

IOResult و FutureResult

لحسن الحظ ، فإن dry-python/returnsالتجريدات الصحيحة موجودة بالفعل.



اسمحوا لي أن أقدم لكم أداة آمنة من النوع ، وصديقة لمايبي ، ومحايد لإطار العمل مكتوبة بالكامل بلغة بايثون. إنه ذو تجريدات رائعة ومريحة ورائعة يمكن استخدامها في أي مشروع على الإطلاق.

خيار متزامن

أولاً ، سنضيف التبعيات للحصول على مثال قابل للتكرار.

pip install returns httpx anyio

بعد ذلك ، دعنا نحول الكود الزائف إلى كود Python يعمل. لنبدأ بالخيار المتزامن.

from typing import Callable
 
import httpx
 
from returns.io import IOResultE, impure_safe
 
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

استغرق الأمر بضعة تغييرات للحصول على رمز العمل:

• IOResultEيعد الاستخدام طريقة وظيفية للتعامل مع أخطاء الإدخال / الإخراج المتزامنة ( لا تعمل الاستثناءات دائمًا ). Resultتسمح لك الأنواع المستندة إلى محاكاة الاستثناءات ، ولكن بقيم منفصلة Failure(). ثم يتم تغليف المخارج الناجحة بنوع Success. عادة لا أحد يهتم بالاستثناءات ، لكننا نفعل ذلك.
• استخدام httpxيمكنه التعامل مع الطلبات المتزامنة وغير المتزامنة.
• استخدم دالة impure_safeلتحويل نوع الإرجاع httpx.getإلى تجريد IOResultE.

خيار غير متزامن

دعنا نحاول أن نفعل الشيء نفسه في التعليمات البرمجية غير المتزامنة.

from typing import Callable
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
 
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

ترى: النتيجة هي نفسها تمامًا ، لكن الكود الآن يعمل بشكل غير متزامن. ومع ذلك ، فإن الجزء الرئيسي منه لم يتغير. ومع ذلك ، عليك الانتباه إلى ما يلي:

• IOResultEتغير في وقت واحد إلى غير متزامن FutureResultE، impure_safe- على future_safe. وهو يعمل نفسه، ولكن إرجاع التجريد مختلفة: FutureResultE.
• تستخدم AsyncClientمن httpx.
• FutureResultيجب تشغيل القيمة الناتجة لأن الوظائف الحمراء لا يمكنها استدعاء نفسها.
• تُستخدم الأداة المساعدة anyioلإظهار أن هذا الأسلوب يعمل مع أي مكتبة غير متزامنة: asyncio، trio، curio.

اثنان في واحد

سأوضح لك كيفية دمج الإصدارات المتزامنة وغير المتزامنة في واجهة برمجة تطبيقات آمنة من النوع.



لم يتم إصدار الأنواع الأعلى من النوع وفئة النوع للعمل مع الإدخال / الإخراج (ستظهر في 0.15.0) ، لذلك سأوضح في المعتاد @overload:

from typing import Callable, Union, overload
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
from returns.io import IOResultE, impure_safe
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    """Sync case."""
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    """Async case."""
 
def fetch_resource_size(
    client_get: Union[
        Callable[[str], IOResultE[httpx.Response]],
        Callable[[str], FutureResultE[httpx.Response]],
    ],
    url: str,
) -> Union[IOResultE[int], FutureResultE[int]]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

نستخدم المصممين @overloadلوصف بيانات الإدخال المسموح بها ونوع القيمة المرتجعة. هل @overloadيمكن قراءة المزيد عن الديكور في بلدي أخرى المادة .



يبدو استدعاء الوظيفة مع عميل متزامن أو غير متزامن كما يلي:

# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>
 
# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

كما ترى ، fetch_resource_sizeفي النسخة المتزامنة ، فإنها تعود IOResultوتنفذها على الفور . بينما في الإصدار غير المتزامن ، يلزم وجود حلقة حدث ، كما هو الحال في coroutine العادي. anyioتستخدم لعرض النتائج. لا يوجد تعليقات



في mypyهذا الكود:

» mypy async_and_sync.py
Success: no issues found in 1 source file

دعونا نرى ماذا يحدث إذا حدث شيء ما.

---lambda response: len(response.content),
+++lambda response: response.content,

mypy يجد بسهولة أخطاء جديدة:

» mypy async_and_sync.py
async_and_sync.py:33: error: Argument 1 to "map" of "IOResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Argument 1 to "map" of "FutureResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Incompatible return value type (got "bytes", expected "int")

خفة اليد وعدم وجود سحر: تتطلب كتابة كود غير متزامن مع التجريدات الصحيحة فقط تكوينًا قديمًا جيدًا. لكن حقيقة حصولنا على نفس واجهة برمجة التطبيقات لأنواع مختلفة رائعة حقًا. على سبيل المثال ، يسمح لك باستخلاص كيفية عمل طلبات HTTP: بشكل متزامن أو غير متزامن.



نأمل أن يكون هذا المثال قد أظهر كيف يمكن أن تكون البرامج غير المتزامنة رائعة حقًا. وإذا جربت الثعبان الجاف / المرتجعات ، فستجد العديد من الأشياء المثيرة للاهتمام. في الإصدار الجديد ، قمنا بالفعل بعمل الأساسيات اللازمة للعمل مع الأنواع العليا وجميع الواجهات اللازمة. يمكن الآن إعادة كتابة الكود أعلاه على النحو التالي:

from typing import Callable, TypeVar

import anyio
import httpx

from returns.future import future_safe
from returns.interfaces.specific.ioresult import IOResultLike2
from returns.io import impure_safe
from returns.primitives.hkt import Kind2, kinded

_IOKind = TypeVar('_IOKind', bound=IOResultLike2)

@kinded
def fetch_resource_size(
    client_get: Callable[[str], Kind2[_IOKind, httpx.Response, Exception]],
    url: str,
) -> Kind2[_IOKind, int, Exception]:
    return client_get(url).map(
        lambda response: len(response.content),
    )


# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

راجع الفرع "الرئيسي" ، وهو يعمل بالفعل هناك.

المزيد من ميزات الثعبان الجاف

فيما يلي بعض الميزات المفيدة الأخرى التي يفخر بها الثعبان الجاف.

• الوظائف المكتوبة partialو@curry .

from returns.curry import curry, partial
 
def example(a: int, b: str) -> float:
    ...
 
reveal_type(partial(example, 1))
# note: Revealed type is 'def (b: builtins.str) -> builtins.float'
 
reveal_type(curry(example))
# note: Revealed type is 'Overload(def (a: builtins.int) -> def (b: builtins.str) -> builtins.float, def (a: builtins.int, b: builtins.str) -> builtins.float)'

يتيح لك هذا @curry، على سبيل المثال ، استخدام مثل هذا:

@curry
def example(a: int, b: str) -> float:
    return float(a + len(b))
 
assert example(1, 'abc') == 4.0
assert example(1)('abc') == 4.0

• خطوط الأنابيب الوظيفية مع نوع الاستدلال.

باستخدام ملحق mypy مخصص ، يمكنك إنشاء خطوط أنابيب وظيفية تُرجع الأنواع.

from returns.pipeline import flow
assert flow(
    [1, 2, 3],
    lambda collection: max(collection),
    lambda max_number: -max_number,
) == -3

عادة ما يكون من غير الملائم العمل مع lambdas في التعليمات البرمجية المكتوبة لأن وسيطاتها تكون دائمًا من النوع Any. الاستدلال mypyيحل هذه المشكلة.



بمساعدتها ، نعرف الآن أي lambda collection: max(collection)نوع Callable[[List[int]], int]، لكن lambda max_number: -max_numberبسيط Callable[[int], int]. في flowيمكن تمرير أي عدد من الحجج، وأنها سوف تعمل بشكل جيد. كل الشكر للبرنامج المساعد.

• RequiresContextFutureResultل حقن التبعية الوظيفية كتبته .

يمكن استخدام التجريد FutureResult، الذي تحدثنا عنه سابقًا ، لتمرير التبعيات صراحة إلى البرامج غير المتزامنة بأسلوب وظيفي.

خطط للمستقبل

قبل أن نصدر أخيرًا الإصدار 1.0 ، علينا حل العديد من المهام المهمة:

• تطبيق الأنواع الأعلى أو مضاهاة ( مشكلة ).
• أضف فئات النوع المناسبة لتنفيذ التجريدات المطلوبة ( المشكلة ).
• ربما جرب مترجمًا mypyc، والذي من المحتمل أن يسمح ببرامج Python المكتوبة المشروحة ليتم تجميعها في ثنائي. ثم dry-python/returnsسيعمل رمز c عدة مرات بشكل أسرع ( مشكلة ).
• اكتشف طرقًا جديدة لكتابة كود وظيفي في بايثون ، مثل "تدوين" .

الاستنتاجات

يمكن أن يحل التركيب والتجريد أي مشكلة. في هذه المقالة ، نظرنا في كيفية حل مشكلة ألوان الوظائف وكتابة كود بسيط وقابل للقراءة ومرن يعمل. وقم بفحص النوع.



جرب Dry-python / Return وانضم إلى أسبوع Python الروسي : في المؤتمر ، سيعقد مطور Dry-python Pablo Aguilar ورشة عمل حول استخدام dry-python لكتابة منطق الأعمال.