returns
Version 0.24.0
関数が意味のある、型指定された、安全なものを返すようにしてください。
mypyでチェックされ、PEP561 と互換性があります。今すぐクイックスタート!
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 -free コードを記述できるコンテナー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 -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 ,
)ずっと良くなりましたね。
多くの開発者は、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が深くネストされていることです。そして、 calculate_pointsや途中にある可能性のある他のすべての関数を含む、コールスタック全体でthresholdを渡す必要があります。それらはすべて、パラメータとして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:最上位レベルを使用することもできます。そしてこのようにして、不要なものを捕まえることになります。このアプローチは、深刻な問題を長期間にわたって私たちから隠す可能性があります。
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] を返します。そして私たちに例外を投げることは決してありません。
また、便利で宣言的な構成のために、フロー関数とバインド関数も使用します。
こうすることで、暗黙的な例外によってコードがランダムな場所で中断されることがなくなります。これで、すべての部分を制御し、明示的なエラーに対する準備が整いました。
この例はまだ完了していません。次の章で改善を続けてみましょう。
この例を別の角度から見てみましょう。そのすべての機能は通常のものと同じように見えます。一見しただけでは、それが純粋であるか不純であるかを区別することは不可能です。
これは非常に重要な結果につながります。つまり、純粋なコードと不純なコードが混在し始めるのです。そんなことはすべきではありません！
これら 2 つの概念が混在すると、テストまたは再利用するときに非常に大きな問題が発生します。デフォルトでは、ほとんどすべてが純粋なはずです。そして、プログラムの不純な部分を明示的にマークする必要があります。
そのため、決して失敗しない不純な関数をマークする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()と同様です。
次に、通常のResult代わりに特別なIOResultタイプを使用する必要があります。違いを見つけてみましょう:
_parse_json関数は、同じ入力に対して常に同じ結果を (うまくいけば) 返します。有効なjsonを解析することも、無効な json で失敗することもできます。そのため、純粋なResultを返します。内部にはIOありません。_make_request関数は不純なので、失敗する可能性があります。インターネット接続がある場合とない場合で、同様のリクエストを 2 つ送信してみてください。同じ入力でも結果は異なります。だからこそ、ここで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 を使用して、純粋な (または「実際の」) 値を取得できます。
このリファクタリング セッションの結果、コードに関するすべてのことがわかりました。
Python のasyncコードにはいくつかの問題があります。
async関数を呼び出すことはできませんawaitステートメントを含む醜い構成FutureコンテナとFutureResultコンテナはこれらの問題を解決します。
Future の主な機能は、同期コンテキストを維持しながら非同期コードを実行できることです。例を見てみましょう。
2 つの関数があり、 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同期コンテキストから非同期関数を操作できます。そして、これら 2 つの領域を混ぜ合わせます。失敗したり例外が発生したりできない操作には、生のFuture使用します。 IOコンテナで行ったロジックとほぼ同じです。
純粋なコードと不純なコードの両方でResultどのように機能するかについてはすでに説明しました。主な考え方は、例外を発生させず、例外を返すということです。単一の例外により、単一のイベントループ内で実行されているすべてのコルーチンが台無しになる可能性があるため、これは非同期コードでは特に重要です。
FutureコンテナとResultコンテナの便利な組み合わせ、 FutureResultがあります。繰り返しますが、これはIOResultとまったく同じですが、不純な非同期コードを対象としています。 HTTP リクエストやファイルシステム操作など、 Future問題が発生する可能性がある場合に使用してください。
乱暴なスロー コルーチンを簡単に穏やかな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使用すると、コードが例外から安全に保たれます。イベントループ内でFutureResultいつでもawaitまたは実行して、同期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 ),
)後で、論理関数を同期してFutureResult返すようにリファクタリングすることもできます。
素敵ですね。
もっと知りたいですか？ドキュメントに行きましょう！または、次の記事をお読みください。
提出する記事はありますか？気軽にプルリクエストを開いてください！
