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 แบบเดียวกับที่เราใช้
ตรวจสอบให้แน่ใจว่าคุณรู้วิธีเริ่มต้นใช้งาน ตรวจสอบเอกสารของเรา! ลองสาธิตของเรา
None - รหัสฟรีasyncdo-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 ) หรือจะใช้คอนเทนเนอร์ Maybe! ประกอบด้วยประเภท 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 -returning ไว้เหนือคอนเทนเนอร์ได้ เพื่อให้บรรลุเป้าหมายนี้ เราจะใช้เมธอด . .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 ,
)ดีขึ้นมากแล้วใช่ไหม?
นักพัฒนาหลายคนใช้ dependency insert บางอย่างใน Python และโดยปกติจะขึ้นอยู่กับแนวคิดที่ว่ามีภาชนะและกระบวนการประกอบบางอย่าง
แนวทางการใช้งานนั้นง่ายกว่ามาก!
ลองจินตนาการว่าคุณมีเกมที่ใช้ 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 ผ่าน callstack ทั้งหมด รวมถึง 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: case ที่จะจับทุกอย่างอย่างแท้จริง และด้วยวิธีนี้เราจะได้จับสิ่งที่ไม่ต้องการได้ แนวทางนี้สามารถซ่อนปัญหาร้ายแรงจากเราได้เป็นเวลานาน
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 มัณฑนากร มันจะส่งคืน Success[YourType] หรือ Failure[Exception] และจะไม่มีวันยกเว้นเรา!
นอกจากนี้เรายังใช้ฟังก์ชันการไหลและการเชื่อมโยงเพื่อการจัดองค์ประกอบที่สะดวกและชัดเจน
วิธีนี้ทำให้เรามั่นใจได้ว่าโค้ดของเราจะไม่แตกในที่สุ่มเนื่องจากมีข้อยกเว้นโดยนัยบางประการ ตอนนี้เราควบคุมทุกส่วนและเตรียมพร้อมสำหรับข้อผิดพลาดที่ชัดเจน
เรายังไม่เสร็จสิ้นกับตัวอย่างนี้ เราจะปรับปรุงต่อไปในบทถัดไป
ลองดูตัวอย่างของเราจากอีกมุมหนึ่ง ฟังก์ชั่นทั้งหมดดูเหมือนปกติ: ไม่สามารถบอกได้ว่าบริสุทธิ์หรือไม่บริสุทธิ์ตั้งแต่แรกเห็น
มันนำไปสู่ผลลัพธ์ที่สำคัญมาก: เราเริ่มผสมโค้ดที่บริสุทธิ์และไม่บริสุทธิ์เข้าด้วยกัน เราไม่ควรทำอย่างนั้น!
เมื่อนำแนวคิดทั้งสองนี้มาผสมกัน เราจะประสบปัญหาอย่างมากเมื่อทำการทดสอบหรือนำกลับมาใช้ใหม่ เกือบทุกอย่างควรจะบริสุทธิ์โดยค่าเริ่มต้น และเราควรทำเครื่องหมายส่วนที่ไม่บริสุทธิ์ของโปรแกรมอย่างชัดเจน
นั่นเป็นเหตุผลที่เราสร้างคอนเทนเนอร์ 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ดังนั้น เพื่อให้เป็นไปตามข้อกำหนดของเราและแยกโค้ดแท้ออกจากโค้ดที่ไม่บริสุทธิ์ เราจึงต้องปรับโครงสร้างตัวอย่างของเราใหม่
มาทำให้ 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 คืออนุญาตให้เรียกใช้โค้ด async ในขณะที่ยังคงรักษาบริบทการซิงค์ไว้ มาดูตัวอย่างกัน
สมมติว่าเรามีสองฟังก์ชัน ฟังก์ชัน 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() เราจะสร้าง coroutine ที่ไม่ต้องรอคอย มันจะไม่คืนค่าที่เราต้องการ
แต่ถ้าเราพยายาม run await first() เราก็จะต้องเปลี่ยน second ให้เป็น async และบางครั้งก็เป็นไปไม่ได้ด้วยเหตุผลหลายประการ
อย่างไรก็ตาม ด้วย Future เราสามารถ "แกล้งทำเป็น" เรียกโค้ด async จากโค้ดซิงค์ได้:
from returns . future import Future
def second () -> Future [ int ]:
return Future ( first ()). map ( lambda num : num + 1 ) โดยไม่ต้องแตะฟังก์ชันอะซิงก์ first ของเราหรือทำอะซิงก์ second เราก็บรรลุเป้าหมายแล้ว ตอนนี้ค่าอะซิงก์ของเราเพิ่มขึ้นภายในฟังก์ชันซิงค์
อย่างไรก็ตาม Future ยังคงต้องดำเนินการภายใน eventloop ที่เหมาะสม:
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 อนุญาตให้คุณทำงานกับฟังก์ชัน async จากบริบทการซิงค์ และนำสองอาณาจักรนี้มารวมกัน ใช้ Raw Future สำหรับการดำเนินการที่ไม่สามารถล้มเหลวหรือเพิ่มข้อยกเว้น เกือบจะเหมือนกับตรรกะเดียวกันกับที่เรามีกับคอนเทนเนอร์ IO ของเรา
เราได้อธิบายไปแล้วว่า Result ทำงานอย่างไรสำหรับโค้ดทั้งโค้ดบริสุทธิ์และโค้ดที่ไม่บริสุทธิ์ แนวคิดหลักคือ: เราไม่ยกข้อยกเว้น แต่เราส่งคืน เป็นสิ่งสำคัญ อย่างยิ่ง ในโค้ดอะซิงก์ เนื่องจากข้อยกเว้นเดียวสามารถทำลายโครูทีนทั้งหมดของเราที่ทำงานอยู่ใน eventloop เดียวได้
เรามีการผสมผสานระหว่างคอนเทนเนอร์ Future และ Result เข้าด้วยกัน: FutureResult นี่เป็นเหมือน IOResult ทุกประการ แต่สำหรับโค้ดอะซิงก์ที่ไม่บริสุทธิ์ ใช้เมื่อ Future ของคุณอาจมีปัญหา เช่น คำขอ HTTP หรือการทำงานของระบบไฟล์
คุณสามารถเปลี่ยน coroutine ที่ขว้างปาอย่างดุเดือดให้กลายเป็น 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 หรือดำเนินการภายใน eventloop 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 ได้
น่ารักใช่มั้ย?
ต้องการมากกว่านี้ไหม? ไปที่เอกสาร! หรืออ่านบทความเหล่านี้:
คุณมีบทความที่จะส่งหรือไม่? อย่าลังเลที่จะเปิดคำขอดึง!
