다운로드 반환 - 소스 코드 다운로드 반환

빌드 상태

함수가 의미 있고 형식화되었으며 안전한 것을 반환하도록 만드세요!

특징

Python 영역에 함수형 프로그래밍을 도입합니다.
선언적 비즈니스 로직을 작성하기 위한 다양한 기본 요소를 제공합니다.
더 나은 아키텍처 적용
주석을 완전히 입력하고 mypy , PEP561 호환으로 확인했습니다.
에뮬레이트된 Higher Kinded Types 지원을 추가합니다.
시행된 법률에 따라 자신만의 데이터 유형을 생성할 수 있는 유형 안전 인터페이스를 제공합니다.
더 나은 구성을 위한 많은 도우미가 있습니다.
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 -free 코드를 작성할 수 있는 컨테이너일 수도 있습니다.
형식화된 기능적 종속성 주입을 사용할 수 있는 RequiresContext 컨테이너
예외를 제거할 수 있는 결과 컨테이너
모든 불순한 작업을 표시하고 구조화하는 IO 컨테이너 및 IOResult
async 코드와 함께 작동하는 Future 컨테이너 및 FutureResult
나만의 컨테이너를 작성해보세요! 자신의 유형에 대한 모든 기능을 계속 사용할 수 있습니다(기존 코드 전체 재사용 및 유형 안전성 포함).
코드를 더 쉽게 만들려면 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 컨테이너를 사용할 수도 있습니다! 이는 각각 기존 상태와 비어 있는( None 대신) 상태를 나타내는 Some 및 Nothing 유형으로 구성됩니다.

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

훨씬 낫지 않나요?

RequireContext 컨테이너

많은 개발자가 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: 최상위 수준으로 갈 수도 있습니다. 그리고 이런 식으로 우리는 원하지 않는 것들을 잡게 될 것입니다. 이 접근 방식은 오랫동안 우리에게 심각한 문제를 숨길 수 있습니다.

파이프 예

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 데코레이터 덕분에 특수 컨테이너 안에 싸인 값을 반환합니다. 성공[YourType] 또는 실패[예외]를 반환합니다. 그리고 결코 우리에게 예외를 던지지 않을 것입니다!

또한 편리하고 선언적인 구성을 위해 흐름 및 바인딩 기능을 사용합니다.

이렇게 하면 암시적인 예외로 인해 코드가 임의의 위치에서 중단되지 않는다는 것을 확신할 수 있습니다. 이제 우리는 모든 부분을 제어하고 명시적인 오류에 대비합니다.

이 예제는 아직 끝나지 않았습니다. 다음 장에서 계속해서 개선해 보겠습니다.

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() 과 같습니다.

그런 다음 일반 Result 대신 특수 IOResult 유형을 사용해야 합니다. 차이점을 찾아봅시다:

_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 와 똑같지만 순수하지 않은 비동기 코드에 대한 것입니다. 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 ),
    )