returns
Version 0.24.0
اجعل وظائفك تُرجع شيئًا ذا معنى ومكتوبًا وآمنًا!
mypy ، وهو متوافق مع PEP561البدء السريع الآن!
pip install returns يمكنك أيضًا تثبيت returns باستخدام أحدث إصدار مدعوم من mypy :
pip install returns[compatible-mypy] ستحتاج أيضًا إلى تكوين البرنامج المساعد mypy الخاص بنا:
# In setup.cfg or mypy.ini:
[mypy]
plugins =
returns.contrib.mypy.returns_pluginأو:
[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ] نوصي أيضًا باستخدام نفس إعدادات mypy التي نستخدمها.
تأكد من أنك تعرف كيفية البدء، وراجع مستنداتنا! جرب العرض التوضيحي الخاص بنا.
Noneasyncdo-notation لتسهيل التعليمات البرمجية الخاصة بك! None يسمى أسوأ خطأ في تاريخ علوم الكمبيوتر.
إذًا، ما الذي يمكننا فعله للتحقق من None في برامجنا؟ يمكنك استخدام النوع الاختياري المدمج وكتابة الكثير من if some is not None: الشروط. ولكن إجراء عمليات فحص null هنا وهناك يجعل التعليمات البرمجية الخاصة بك غير قابلة للقراءة .
user : Optional [ User ]
discount_program : Optional [ 'DiscountProgram' ] = None
if user is not None :
balance = user . get_balance ()
if balance is not None :
credit = balance . credit_amount ()
if credit is not None and credit > 0 :
discount_program = choose_discount ( credit ) أو يمكنك استخدام ربما الحاوية! وهو يتألف من نوعين Some و Nothing ، يمثلان الحالة الحالية والحالة الفارغة (بدلاً من None ) على التوالي.
from typing import Optional
from returns . maybe import Maybe , maybe
@ maybe # decorator to convert existing Optional[int] to Maybe[int]
def bad_function () -> Optional [ int ]:
...
maybe_number : Maybe [ float ] = bad_function (). bind_optional (
lambda number : number / 2 ,
)
# => Maybe will return Some[float] only if there's a non-None value
# Otherwise, will return Nothing يمكنك التأكد من أنه لن يتم استدعاء الأسلوب .bind_optional() لـ Nothing . انسَ الأخطاء المتعلقة None إلى الأبد!
يمكننا أيضًا ربط وظيفة إرجاع Optional عبر حاوية. لتحقيق ذلك، سنستخدم طريقة .bind_optional .
وإليك كيف سيبدو الكود الأولي المُعاد هيكلته:
user : Optional [ User ]
# Type hint here is optional, it only helps the reader here:
discount_program : Maybe [ 'DiscountProgram' ] = Maybe . from_optional (
user ,
). bind_optional ( # This won't be called if `user is None`
lambda real_user : real_user . get_balance (),
). bind_optional ( # This won't be called if `real_user.get_balance()` is None
lambda balance : balance . credit_amount (),
). bind_optional ( # And so on!
lambda credit : choose_discount ( credit ) if credit > 0 else None ,
)أفضل بكثير، أليس كذلك؟
يستخدم العديد من المطورين نوعًا من حقن التبعية في بايثون. وعادة ما يعتمد ذلك على فكرة وجود نوع من الحاوية وعملية التجميع.
النهج الوظيفي هو أبسط من ذلك بكثير!
تخيل أن لديك لعبة تعتمد على django ، حيث تمنح المستخدمين نقاطًا مقابل كل حرف محزر في الكلمة (يتم وضع علامة على الأحرف غير المخمنة كـ '.' ):
from django . http import HttpRequest , HttpResponse
from words_app . logic import calculate_points
def view ( request : HttpRequest ) -> HttpResponse :
user_word : str = request . POST [ 'word' ] # just an example
points = calculate_points ( user_word )
... # later you show the result to user somehow
# Somewhere in your `words_app/logic.py`:
def calculate_points ( word : str ) -> int :
guessed_letters_count = len ([ letter for letter in word if letter != '.' ])
return _award_points_for_letters ( guessed_letters_count )
def _award_points_for_letters ( guessed : int ) -> int :
return 0 if guessed < 5 else guessed # minimum 6 points possible!مذهل! إنه يعمل، والمستخدمون سعداء، ومنطقك نقي ورائع. ولكن، قررت لاحقًا أن تجعل اللعبة أكثر متعة: فلنجعل الحد الأدنى من الحروف المسؤولة قابلاً للتكوين من أجل تحدي إضافي.
يمكنك فقط القيام بذلك مباشرة:
def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
return 0 if guessed < threshold else guessed المشكلة هي أن _award_points_for_letters متداخلة بعمق. وبعد ذلك يتعين عليك تجاوز threshold عبر مجموعة الاستدعاءات بأكملها، بما في ذلك calculate_points وجميع الوظائف الأخرى التي قد تكون في الطريق. سيتعين عليهم جميعًا قبول threshold كمعلمة! وهذا غير مفيد على الإطلاق! ستواجه قواعد الأكواد الكبيرة صعوبة كبيرة بسبب هذا التغيير.
حسنًا، يمكنك استخدام django.settings (أو ما شابه) مباشرة في وظيفة _award_points_for_letters الخاصة بك. ودمر منطقك النقي بتفاصيل محددة للإطار . هذا قبيح!
أو يمكنك استخدام حاوية RequiresContext . دعونا نرى كيف يتغير الكود الخاص بنا:
from django . conf import settings
from django . http import HttpRequest , HttpResponse
from words_app . logic import calculate_points
def view ( request : HttpRequest ) -> HttpResponse :
user_word : str = request . POST [ 'word' ] # just an example
points = calculate_points ( user_words )( settings ) # passing the dependencies
... # later you show the result to user somehow
# Somewhere in your `words_app/logic.py`:
from typing import Protocol
from returns . context import RequiresContext
class _Deps ( Protocol ): # we rely on abstractions, not direct values or types
WORD_THRESHOLD : int
def calculate_points ( word : str ) -> RequiresContext [ int , _Deps ]:
guessed_letters_count = len ([ letter for letter in word if letter != '.' ])
return _award_points_for_letters ( guessed_letters_count )
def _award_points_for_letters ( guessed : int ) -> RequiresContext [ int , _Deps ]:
return RequiresContext (
lambda deps : 0 if guessed < deps . WORD_THRESHOLD else guessed ,
) والآن يمكنك تمرير تبعياتك بطريقة مباشرة وصريحة حقًا. واحصل على نوع الأمان للتحقق مما تمر به لتغطية ظهرك. تحقق من مستندات RequiresContext للمزيد. هناك سوف تتعلم كيفية صنع '.' شكلي أيضا.
لدينا أيضًا RequiresContextResult للعمليات المرتبطة بالسياق والتي قد تفشل. وأيضًا RequiresContextIOResult وRequiresContextFutureResult.
من فضلك، تأكد من أنك على دراية أيضًا بالبرمجة الموجهة للسكك الحديدية.
ضع في اعتبارك هذا الرمز الذي يمكنك العثور عليه في أي مشروع python .
import requests
def fetch_user_profile ( user_id : int ) -> 'UserProfile' :
"""Fetches UserProfile dict from foreign API."""
response = requests . get ( '/api/users/{0}' . format ( user_id ))
response . raise_for_status ()
return response . json () يبدو شرعيا، أليس كذلك؟ يبدو أيضًا وكأنه رمز بسيط جدًا للاختبار. كل ما تحتاجه هو السخرية من requests.get لإعادة البنية التي تحتاجها.
ولكن هناك مشكلات مخفية في نموذج التعليمات البرمجية الصغير هذا يكاد يكون من المستحيل اكتشافه للوهلة الأولى.
دعونا نلقي نظرة على نفس الكود بالضبط، ولكن مع شرح جميع المشاكل المخفية.
import requests
def fetch_user_profile ( user_id : int ) -> 'UserProfile' :
"""Fetches UserProfile dict from foreign API."""
response = requests . get ( '/api/users/{0}' . format ( user_id ))
# What if we try to find user that does not exist?
# Or network will go down? Or the server will return 500?
# In this case the next line will fail with an exception.
# We need to handle all possible errors in this function
# and do not return corrupt data to consumers.
response . raise_for_status ()
# What if we have received invalid JSON?
# Next line will raise an exception!
return response . json ()الآن، أصبحت جميع المشاكل (وربما جميعها؟) واضحة. كيف يمكننا التأكد من أن هذه الوظيفة ستكون آمنة للاستخدام داخل منطق أعمالنا المعقد؟
نحن حقا لا نستطيع أن نكون متأكدين! سيتعين علينا إنشاء الكثير من حالات try except فقط للقبض على الاستثناءات المتوقعة. سيصبح الكود الخاص بنا معقدًا وغير قابل للقراءة مع كل هذه الفوضى!
أو يمكننا الانتقال إلى المستوى الأعلى except Exception: الحالة لالتقاط كل شيء حرفيًا. وبهذه الطريقة سننتهي بالقبض على الأشخاص غير المرغوب فيهم. هذا النهج يمكن أن يخفي عنا مشاكل خطيرة لفترة طويلة.
import requests
from returns . result import Result , safe
from returns . pipeline import flow
from returns . pointfree import bind
def fetch_user_profile ( user_id : int ) -> Result [ 'UserProfile' , Exception ]:
"""Fetches `UserProfile` TypedDict from foreign API."""
return flow (
user_id ,
_make_request ,
bind ( _parse_json ),
)
@ safe
def _make_request ( user_id : int ) -> requests . Response :
# TODO: we are not yet done with this example, read more about `IO`:
response = requests . get ( '/api/users/{0}' . format ( user_id ))
response . raise_for_status ()
return response
@ safe
def _parse_json ( response : requests . Response ) -> 'UserProfile' :
return response . json ()الآن لدينا طريقة نظيفة وآمنة ومعلنة للتعبير عن احتياجات أعمالنا:
الآن، بدلاً من إرجاع القيم العادية، نعيد القيم المغلفة داخل حاوية خاصة بفضل الديكور @safe. سيُرجع النجاح [YourType] أو الفشل [الاستثناء]. ولن يرمي استثناء علينا!
نستخدم أيضًا وظائف التدفق والربط لتكوين مفيد وتصريحي.
بهذه الطريقة يمكننا التأكد من أن الكود الخاص بنا لن ينكسر في أماكن عشوائية بسبب بعض الاستثناءات الضمنية. الآن نحن نتحكم في جميع الأجزاء ونستعد للأخطاء الصريحة.
لم ننته بعد من هذا المثال، فلنواصل تحسينه في الفصل التالي.
دعونا ننظر إلى مثالنا من زاوية أخرى. تبدو جميع وظائفها وكأنها وظائف عادية: فمن المستحيل معرفة ما إذا كانت طاهرة أم نجسة من النظرة الأولى.
إنه يؤدي إلى نتيجة مهمة للغاية: نبدأ في مزج التعليمات البرمجية النقية وغير النقية معًا . لا ينبغي لنا أن نفعل ذلك!
عندما يتم الخلط بين هذين المفهومين فإننا نعاني بشدة عند اختباره أو إعادة استخدامه. يجب أن يكون كل شيء تقريبًا نقيًا بشكل افتراضي. ويجب علينا أن نحدد بشكل صريح الأجزاء غير النقية من البرنامج.
لهذا السبب قمنا بإنشاء حاوية IO لتمييز الوظائف غير النقية التي لا تفشل أبدًا.
تستخدم هذه الوظائف غير النقية تاريخًا random أو حاليًا أو بيئة أو وحدة تحكم:
import random
import datetime as dt
from returns . io import IO
def get_random_number () -> IO [ int ]: # or use `@impure` decorator
return IO ( random . randint ( 1 , 10 )) # isn't pure, because random
now : Callable [[], IO [ dt . datetime ]] = impure ( dt . datetime . now )
@ impure
def return_and_show_next_number ( previous : int ) -> int :
next_number = previous + 1
print ( next_number ) # isn't pure, because does IO
return next_numberالآن يمكننا أن نرى بوضوح أي الوظائف نقية وأيها غير نقية. يساعدنا هذا كثيرًا في إنشاء تطبيقات كبيرة، واختبار الوحدة للتعليمات البرمجية الخاصة بك، وتأليف منطق الأعمال معًا.
كما قلنا سابقًا، نستخدم IO عندما نتعامل مع وظائف لا تفشل.
ماذا لو كانت وظيفتنا قد تفشل وأصبحت غير نقية؟ مثل requests.get() الذي ذكرناه سابقًا في مثالنا.
ثم يتعين علينا استخدام نوع IOResult خاص بدلاً من Result العادية. دعونا نجد الفرق:
_parse_json الخاصة بنا دائمًا بإرجاع نفس النتيجة (نأمل) لنفس الإدخال: يمكنك إما تحليل json صالح أو الفشل في واحد غير صالح. لهذا السبب نعيد Result النقية، لا يوجد IO بالداخل_make_request لدينا غير نقية ويمكن أن تفشل. حاول إرسال طلبين مماثلين مع وبدون اتصال بالإنترنت. ستكون النتيجة مختلفة لنفس الإدخال. لهذا السبب يجب علينا استخدام IOResult هنا: يمكن أن يفشل ويحتوي على IOلذا، من أجل تلبية متطلباتنا وفصل التعليمات البرمجية النقية عن التعليمات البرمجية غير النقية، يتعين علينا إعادة هيكلة مثالنا.
دعونا نجعل عملية الإدخال/الإخراج واضحة!
import requests
from returns . io import IOResult , impure_safe
from returns . result import safe
from returns . pipeline import flow
from returns . pointfree import bind_result
def fetch_user_profile ( user_id : int ) -> IOResult [ 'UserProfile' , Exception ]:
"""Fetches `UserProfile` TypedDict from foreign API."""
return flow (
user_id ,
_make_request ,
# before: def (Response) -> UserProfile
# after safe: def (Response) -> ResultE[UserProfile]
# after bind_result: def (IOResultE[Response]) -> IOResultE[UserProfile]
bind_result ( _parse_json ),
)
@ impure_safe
def _make_request ( user_id : int ) -> requests . Response :
response = requests . get ( '/api/users/{0}' . format ( user_id ))
response . raise_for_status ()
return response
@ safe
def _parse_json ( response : requests . Response ) -> 'UserProfile' :
return response . json ()وبعد ذلك يمكننا استخدام unsafe_perform_io في مكان ما في المستوى الأعلى لبرنامجنا للحصول على القيمة النقية (أو "الحقيقية").
نتيجة لجلسة إعادة البناء هذه، أصبحنا نعرف كل شيء عن الكود الخاص بنا:
هناك العديد من المشكلات المتعلقة بالكود async في Python:
async من وظيفة متزامنةawait حاويات Future و FutureResult تحل هذه المشكلات!
الميزة الرئيسية لبرنامج Future هي أنه يسمح بتشغيل التعليمات البرمجية غير المتزامنة مع الحفاظ على سياق المزامنة. دعونا نرى مثالا.
لنفترض أن لدينا دالتين، first تُرجع رقمًا second تزيده:
async def first () -> int :
return 1
def second (): # How can we call `first()` from here?
return first () + 1 # Boom! Don't do this. We illustrate a problem here. إذا حاولنا تشغيل first() فقط، فسنقوم بإنشاء كوروتين غير منتظر. ولن يعيد القيمة التي نريدها.
لكن، إذا حاولنا تشغيل await first() ، فسنحتاج إلى تغيير second ليكون async . وأحيانا لا يكون ذلك ممكنا لأسباب مختلفة.
ومع ذلك، مع Future يمكننا "التظاهر" باستدعاء رمز غير متزامن من رمز المزامنة:
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 ) دون لمس وظيفة المزامنة first لدينا أو إجراء المزامنة second ، فقد حققنا هدفنا. الآن، تتم زيادة قيمة المزامنة لدينا داخل وظيفة المزامنة.
ومع ذلك، لا يزال يتطلب تنفيذ Future داخل حلقة حدث مناسبة:
import anyio # or asyncio, or any other lib
# We can then pass our `Future` to any library: asyncio, trio, curio.
# And use any event loop: regular, uvloop, even a custom one, etc
assert anyio . run ( second (). awaitable ) == 2 كما ترون، يسمح لك Future بالعمل مع وظائف غير متزامنة من سياق المزامنة. ولخلط هذين المجالين معًا. استخدم Future الخام للعمليات التي لا يمكن أن تفشل أو تثير الاستثناءات. إلى حد كبير نفس المنطق الذي استخدمناه مع حاوية IO الخاصة بنا.
لقد قمنا بالفعل بتغطية كيفية عمل Result لكل من التعليمات البرمجية النقية وغير النقية. الفكرة الرئيسية هي: نحن لا نطرح الاستثناءات، بل نعيدها. إنه أمر بالغ الأهمية بشكل خاص في التعليمات البرمجية غير المتزامنة، لأن استثناء واحد يمكن أن يدمر جميع coroutines لدينا التي تعمل في حلقة حدث واحدة.
لدينا مجموعة سهلة الاستخدام من حاويات Future و Result : FutureResult . مرة أخرى، هذا يشبه IOResult تمامًا، لكن بالنسبة للتعليمات البرمجية غير المتزامنة غير النقية. استخدمه عندما يواجه Future مشكلات: مثل طلبات HTTP أو عمليات نظام الملفات.
يمكنك بسهولة تحويل أي كوروتين بري إلى FutureResult هادئة:
import anyio
from returns . future import future_safe
from returns . io import IOFailure
@ future_safe
async def raising ():
raise ValueError ( 'Not so fast!' )
ioresult = anyio . run ( raising . awaitable ) # all `Future`s return IO containers
assert ioresult == IOFailure ( ValueError ( 'Not so fast!' )) # True سيؤدي استخدام FutureResult إلى الحفاظ على التعليمات البرمجية الخاصة بك في مأمن من الاستثناءات. يمكنك دائمًا await أو تنفيذ أي FutureResult داخل حلقة الأحداث للحصول على مثيل IOResult متزامن للعمل معه بطريقة متزامنة.
في السابق، كان عليك await كثيرًا أثناء كتابة التعليمات البرمجية async :
async def fetch_user ( user_id : int ) -> 'User' :
...
async def get_user_permissions ( user : 'User' ) -> 'Permissions' :
...
async def ensure_allowed ( permissions : 'Permissions' ) -> bool :
...
async def main ( user_id : int ) -> bool :
# Also, don't forget to handle all possible errors with `try / except`!
user = await fetch_user ( user_id ) # We will await each time we use a coro!
permissions = await get_user_permissions ( user )
return await ensure_allowed ( permissions )بعض الناس موافقون على ذلك، لكن بعض الناس لا يحبون هذا الأسلوب الحتمي. المشكلة هي أنه لم يكن هناك خيار.
ولكن الآن، يمكنك أن تفعل الشيء نفسه بأسلوب وظيفي! بمساعدة حاويات Future و FutureResult :
import anyio
from returns . future import FutureResultE , future_safe
from returns . io import IOSuccess , IOFailure
@ future_safe
async def fetch_user ( user_id : int ) -> 'User' :
...
@ future_safe
async def get_user_permissions ( user : 'User' ) -> 'Permissions' :
...
@ future_safe
async def ensure_allowed ( permissions : 'Permissions' ) -> bool :
...
def main ( user_id : int ) -> FutureResultE [ bool ]:
# We can now turn `main` into a sync function, it does not `await` at all.
# We also don't care about exceptions anymore, they are already handled.
return fetch_user ( user_id ). bind ( get_user_permissions ). bind ( ensure_allowed )
correct_user_id : int # has required permissions
banned_user_id : int # does not have required permissions
wrong_user_id : int # does not exist
# We can have correct business results:
assert anyio . run ( main ( correct_user_id ). awaitable ) == IOSuccess ( True )
assert anyio . run ( main ( banned_user_id ). awaitable ) == IOSuccess ( False )
# Or we can have errors along the way:
assert anyio . run ( main ( wrong_user_id ). awaitable ) == IOFailure (
UserDoesNotExistError (...),
)أو حتى شيء فاخر حقًا:
from returns . pointfree import bind
from returns . pipeline import flow
def main ( user_id : int ) -> FutureResultE [ bool ]:
return flow (
fetch_user ( user_id ),
bind ( get_user_permissions ),
bind ( ensure_allowed ),
) يمكننا لاحقًا أيضًا إعادة هيكلة وظائفنا المنطقية لتتم مزامنتها وإرجاع FutureResult .
جميل، أليس كذلك؟
هل تريد المزيد؟ اذهب إلى المستندات! أو إقرأ هذه المقالات:
هل لديك مقال لتقديمه؟ لا تتردد في فتح طلب السحب!
