返回下載 - 返回原始碼下載

建置狀態

讓你的函數回傳一些有意義的、有類型的、安全的東西！

特徵

將函數式程式設計引入 Python 領域
提供一堆原語來編寫聲明性業務邏輯
實作更好的架構
完全鍵入註解並使用mypy檢查，相容於 PEP561
添加模擬的更高種類類型支持
提供類型安全的介面來建立您自己的具有強制法律的資料類型
有一堆幫助者可以更好地構圖
Pythonic 寫起來和讀起來都很愉快？
支援函數和協程，與框架無關
易於入門：有大量文件、測試和教程

立即快速啟動！

安裝

pip install returns

您也可以使用最新支援的mypy版本安裝returns ：

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自由程式碼
RequiresContext 容器，允許您使用類型化函數依賴注入
結果容器可讓您擺脫異常
IO容器和IOResult標記所有不純的操作並建構它們
Future 容器和 FutureResult 使用async程式碼
編寫你自己的容器！您仍然擁有自己類型的所有功能（包括完整的現有程式碼重複使用和類型安全）
使用do-notation使您的程式碼更容易！

也許是容器

None被稱為計算機科學史上最嚴重的錯誤。

那麼，我們可以做什麼來檢查程式中是否存在None ？您可以使用內建的Optional類型並編寫很多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

您可以確定不會為Nothing呼叫.bind_optional()方法。永遠忘記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 ,
)

好多了，不是嗎？

RequiresContext 容器

許多開發人員確實在 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 ，包括calculate_points和可能正在路上的所有其他函數。他們所有人都必須接受threshold作為參數！這根本沒用！大型程式碼庫將因這項變更而陷入困境。

好的，您可以直接在_award_points_for_letters函數中使用django.settings （或類似的）。並用框架特定的細節破壞你的純粹邏輯。太醜了！

或者您可以使用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 之外的頂層來捕捉幾乎所有內容。這樣我們最終就會捕捉不需要的東西。這種方法可以長期地向我們隱藏嚴重的問題。

管道範例

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()">

 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容器

讓我們從另一個角度來看我們的例子。它的所有功能看起來都和正常的一樣，乍一看無法分辨它們是純粹的還是不純粹的。

這導致了一個非常重要的後果：我們開始將純程式碼和不純程式碼混合在一起。我們不應該這樣做！

當這兩個概念混合在一起時，我們在測試或重複使用它時會遭受非常嚴重的痛苦。預設情況下，幾乎所有內容都應該是純的。我們應該明確標記程序中不純淨的部分。

這就是為什麼我們創建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

正如已經說過的，當我們處理不會失敗的函數時，我們會使用IO 。

如果我們的函數可能失敗且不純淨怎麼辦？就像我們前面的範例中的requests.get() 。

然後我們必須使用特殊的IOResult類型而不是常規的Result 。我們來找出差異：

對於相同的輸入，我們的_parse_json函數總是傳回相同的結果（希望如此）：您可以解析有效的json ，也可以解析無效的 json。這就是為什麼我們回傳純Result ，裡面沒有IO
我們的_make_request函數是不純粹的並且可能會失敗。嘗試在有網路連線和沒有網路連線的情況下發送兩個類似的請求。對於相同的輸入，結果會有所不同。這就是為什麼我們必須在這裡使用IOResult ：它可能失敗並且有IO

因此，為了滿足我們的要求並將純程式碼與不純程式碼分開，我們必須重構我們的範例。

顯式IO

讓我們明確一下我們的 IO！

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()">

 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 來取得純（或「真實」）值。

透過這次重構，我們了解了有關程式碼的一切：

哪些部件可能會故障，
哪些部分不純，
如何以智慧、可讀且類型安全的方式組合它們。

未來容器

Python 中的async程式碼存在幾個問題：

您無法從同步函數呼叫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如何適用於純程式碼和非純程式碼。主要想法是：我們不引發異常，而是回傳異常。它在非同步程式碼中尤其重要，因為單一異常可能會破壞在單一事件循環中運行的所有協程。

我們有一個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實例，以同步方式使用它。

更好的非同步組合

以前，您在編寫非async程式碼時必須執行大量await操作：

 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 ),
    )