mengembalikan Unduhan - mengembalikan unduhan kode sumber

Membangun Status

Jadikan fungsi Anda mengembalikan sesuatu yang bermakna, terketik, dan aman!

Fitur

Membawa pemrograman fungsional ke dunia Python
Menyediakan banyak primitif untuk menulis logika bisnis deklaratif
Menerapkan arsitektur yang lebih baik
Diketik sepenuhnya dengan anotasi dan diperiksa dengan mypy , kompatibel dengan PEP561
Menambahkan dukungan Tiruan Tinggi yang ditiru
Menyediakan antarmuka tipe-aman untuk membuat tipe data Anda sendiri dengan hukum yang ditegakkan
Memiliki banyak pembantu untuk komposisi yang lebih baik
Pythonic dan menyenangkan untuk ditulis dan dibaca?
Fungsi dukungan dan coroutine, kerangka kerja agnostik
Mudah untuk memulai: memiliki banyak dokumen, tes, dan tutorial

Mulai cepat sekarang!

Instalasi

pip install returns

Anda juga dapat menginstal returns dengan versi mypy terbaru yang didukung:

pip install returns[compatible-mypy]

Anda juga perlu mengkonfigurasi plugin mypy kami:

 # In setup.cfg or mypy.ini:
[mypy]
plugins =
  returns.contrib.mypy.returns_plugin

atau:

[ tool . mypy ]
plugins = [ " returns.contrib.mypy.returns_plugin " ]

Kami juga menyarankan untuk menggunakan pengaturan mypy yang sama dengan yang kami gunakan.

Pastikan Anda tahu cara memulainya, lihat dokumen kami! Coba demo kami.

Isi

Mungkin wadah yang memungkinkan Anda menulis kode None -bebas
Kontainer RequiresContext yang memungkinkan Anda menggunakan injeksi ketergantungan fungsional yang diketik
Wadah hasil yang memungkinkan Anda menghilangkan pengecualian
Kontainer IO dan IOResult yang menandai semua operasi tidak murni dan menyusunnya
Penampung masa depan dan FutureResult berfungsi dengan kode async
Tulis wadah Anda sendiri! Anda akan tetap memiliki semua fitur untuk tipe Anda sendiri (termasuk penggunaan kembali kode lengkap dan keamanan tipe)
Gunakan do-notation untuk membuat kode Anda lebih mudah!

Mungkin wadah

None yang disebut sebagai kesalahan terburuk dalam sejarah Ilmu Komputer.

Jadi, apa yang bisa kita lakukan untuk memeriksa None di program kita? Anda dapat menggunakan tipe Opsional bawaan dan menulis banyak kondisi if some is not None: :. Namun, melakukan pemeriksaan null di sana-sini membuat kode Anda tidak dapat dibaca .

 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 )

Atau Anda bisa menggunakan wadah Maybe! Ini terdiri dari tipe Some dan Nothing , masing-masing mewakili status yang ada dan status kosong (bukan 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

Anda dapat yakin bahwa metode .bind_optional() tidak akan dipanggil untuk Nothing . Lupakan kesalahan terkait None selamanya!

Kita juga dapat mengikat fungsi Optional -returning pada sebuah container. Untuk mencapai ini, kita akan menggunakan metode .bind_optional .

Dan inilah tampilan kode awal Anda yang telah difaktorkan ulang:

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

Jauh lebih baik, bukan?

Kontainer RequiresContext

Banyak pengembang menggunakan semacam injeksi ketergantungan dengan Python. Dan biasanya didasarkan pada gagasan bahwa ada semacam wadah dan proses perakitan.

Pendekatan fungsional jauh lebih sederhana!

Bayangkan bahwa Anda mempunyai permainan berbasis django , di mana Anda memberi penghargaan kepada pengguna dengan poin untuk setiap huruf yang ditebak dalam sebuah kata (huruf yang tidak dapat ditebak ditandai sebagai '.' ):

 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!

Luar biasa! Berhasil, pengguna senang, logika Anda murni dan mengagumkan. Namun, nanti Anda memutuskan untuk membuat permainan lebih menyenangkan: mari buat ambang batas huruf minimal yang dapat dipertanggungjawabkan dapat dikonfigurasi untuk tantangan ekstra.

Anda bisa melakukannya secara langsung:

 def _award_points_for_letters ( guessed : int , threshold : int ) -> int :
    return 0 if guessed < threshold else guessed

Masalahnya adalah _award_points_for_letters sangat bersarang. Dan kemudian Anda harus melewati threshold melalui seluruh tumpukan panggilan, termasuk calculate_points dan semua fungsi lain yang mungkin sedang diproses. Semuanya harus menerima threshold sebagai parameter! Ini sama sekali tidak berguna! Basis kode yang besar akan mengalami kesulitan akibat perubahan ini.

Oke, Anda bisa langsung menggunakan django.settings (atau serupa) di fungsi _award_points_for_letters Anda. Dan hancurkan logika murni Anda dengan detail spesifik kerangka kerja . Itu jelek!

Atau Anda dapat menggunakan wadah RequiresContext . Mari kita lihat bagaimana kode kita berubah:

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

Dan sekarang Anda dapat meneruskan dependensi Anda dengan cara yang sangat langsung dan eksplisit. Dan miliki keamanan tipe untuk memeriksa apa yang Anda lewati untuk menutupi punggung Anda. Lihat dokumen RequiresContext untuk mengetahui lebih lanjut. Di sana Anda akan belajar cara membuat '.' juga dapat dikonfigurasi.

Kami juga memiliki RequiresContextResult untuk operasi terkait konteks yang mungkin gagal. Dan juga RequiresContextIOResult dan RequiresContextFutureResult.

Wadah hasil

Harap pastikan bahwa Anda juga mengetahui Pemrograman Berorientasi Kereta Api.

Pendekatan langsung

Pertimbangkan kode ini yang dapat Anda temukan di proyek python mana pun .

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

Tampaknya sah, bukan? Tampaknya ini juga merupakan kode yang cukup mudah untuk diuji. Yang Anda butuhkan hanyalah meniru requests.get Dapatkan untuk mengembalikan struktur yang Anda butuhkan.

Namun, ada masalah tersembunyi dalam contoh kode kecil ini yang hampir mustahil ditemukan pada pandangan pertama.

Masalah tersembunyi

Mari kita lihat kode yang sama persis, tetapi dengan semua masalah tersembunyi yang dijelaskan.

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

Sekarang, semua (mungkin semua?) masalah sudah jelas. Bagaimana kita bisa yakin bahwa fungsi ini aman digunakan dalam logika bisnis kita yang kompleks?

Kami benar-benar tidak yakin! Kita harus membuat banyak kasus try dan except hanya untuk menangkap pengecualian yang diharapkan. Kode kita akan menjadi rumit dan tidak dapat dibaca dengan semua kekacauan ini!

Atau kita bisa menggunakan level teratas except Exception: case untuk menangkap semuanya secara harfiah. Dan dengan cara ini kita akan menangkap orang-orang yang tidak diinginkan. Pendekatan ini dapat menyembunyikan masalah serius dari kita untuk waktu yang lama.

Contoh pipa

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

Sekarang kami memiliki cara yang bersih dan aman serta deklaratif untuk mengekspresikan kebutuhan bisnis kami:

Kita mulai dari membuat permintaan, yang sewaktu-waktu bisa saja gagal,
Kemudian parsing respon jika permintaan berhasil,
Dan kemudian kembalikan hasilnya.

Sekarang, alih-alih mengembalikan nilai reguler, kami mengembalikan nilai yang dibungkus dalam wadah khusus berkat dekorator @safe. Ini akan mengembalikan Sukses[Jenis Anda] atau Kegagalan[Pengecualian]. Dan tidak akan pernah memberikan pengecualian pada kami!

Kami juga menggunakan fungsi aliran dan pengikatan untuk komposisi praktis dan deklaratif.

Dengan cara ini kita dapat yakin bahwa kode kita tidak akan rusak di tempat acak karena beberapa pengecualian implisit. Sekarang kami mengontrol semua bagian dan bersiap menghadapi kesalahan yang jelas.

Kita belum selesai dengan contoh ini, mari kita terus memperbaikinya di bab berikutnya.

wadah IO

Mari kita lihat contoh kita dari sudut lain. Semua fungsinya terlihat seperti fungsi biasa: tidak mungkin untuk mengetahui apakah murni atau tidak murni pada pandangan pertama.

Ini membawa konsekuensi yang sangat penting: kita mulai mencampurkan kode murni dan tidak murni . Kita seharusnya tidak melakukan itu!

Jika kedua konsep ini digabungkan, kami akan mengalami kesulitan saat menguji atau menggunakannya kembali. Hampir semuanya harus murni secara default. Dan kita harus secara eksplisit menandai bagian program yang tidak murni.

Itu sebabnya kami membuat wadah IO untuk menandai fungsi tidak murni yang tidak pernah gagal.

Fungsi-fungsi tidak murni ini menggunakan random , datetime saat ini, lingkungan, atau konsol:

 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

Sekarang kita bisa melihat dengan jelas fungsi mana yang murni dan mana yang tidak murni. Ini sangat membantu kami dalam membangun aplikasi besar, menguji unit kode Anda, dan menyusun logika bisnis bersama-sama.

IO yang merepotkan

Seperti yang telah dikatakan, kami menggunakan IO ketika kami menangani fungsi yang tidak gagal.

Bagaimana jika fungsi kita gagal dan tidak murni? Seperti requests.get() yang kita miliki sebelumnya dalam contoh kita.

Kemudian kita harus menggunakan tipe IOResult khusus daripada Result biasa. Mari temukan perbedaannya:

Fungsi _parse_json kami selalu mengembalikan hasil yang sama (mudah-mudahan) untuk masukan yang sama: Anda dapat mengurai json yang valid atau gagal pada json yang tidak valid. Itu sebabnya kami mengembalikan Result murni, tidak ada IO di dalamnya
Fungsi _make_request kami tidak murni dan bisa gagal. Cobalah untuk mengirim dua permintaan serupa dengan dan tanpa koneksi internet. Hasilnya akan berbeda untuk masukan yang sama. Itu sebabnya kita harus menggunakan IOResult di sini: bisa gagal dan memiliki IO

Jadi, untuk memenuhi persyaratan kita dan memisahkan kode murni dari kode tidak murni, kita harus memfaktorkan ulang contoh kita.

IO eksplisit

Mari buat IO kita eksplisit!

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

Dan nanti kita bisa menggunakan unsafe_perform_io di suatu tempat di tingkat atas program kita untuk mendapatkan nilai murni (atau "nyata").

Sebagai hasil dari sesi pemfaktoran ulang ini, kami mengetahui segalanya tentang kode kami:

Bagian mana yang bisa gagal,
Bagian mana yang najis,
Cara menyusunnya dengan cara yang cerdas, mudah dibaca, dan aman untuk mengetik.

Wadah masa depan

Ada beberapa masalah dengan kode async di Python:

Anda tidak dapat memanggil fungsi async dari fungsi sinkronisasi
Pengecualian apa pun yang dilontarkan secara tidak terduga dapat merusak seluruh loop acara Anda
Komposisi jelek dengan banyak pernyataan await

Kontainer Future dan FutureResult memecahkan masalah ini!

Mencampur kode sinkronisasi dan asinkron

Fitur utama Future adalah memungkinkan untuk menjalankan kode async sambil mempertahankan konteks sinkronisasi. Mari kita lihat sebuah contoh.

Katakanlah kita memiliki dua fungsi, fungsi first mengembalikan angka dan fungsi second menambahnya:

 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.

Jika kita mencoba menjalankan first() , kita hanya akan membuat coroutine yang tidak ditunggu. Itu tidak akan mengembalikan nilai yang kita inginkan.

Namun, jika kita mencoba menjalankan await first() , maka kita perlu mengubah second menjadi async . Dan terkadang hal itu tidak bisa dilakukan karena berbagai alasan.

Namun, dengan Future kita dapat "berpura-pura" memanggil kode async dari kode sinkronisasi:

 from returns . future import Future

def second () -> Future [ int ]:
    return Future ( first ()). map ( lambda num : num + 1 )

Tanpa menyentuh fungsi async first atau membuat async second , kita telah mencapai tujuan kita. Sekarang, nilai async kita bertambah di dalam fungsi sinkronisasi.

Namun, Future masih perlu dieksekusi di dalam eventloop yang tepat:

 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

Seperti yang Anda lihat, Future memungkinkan Anda bekerja dengan fungsi async dari konteks sinkronisasi. Dan untuk menggabungkan kedua bidang ini menjadi satu. Gunakan Future mentah untuk operasi yang tidak boleh gagal atau memunculkan pengecualian. Logika yang hampir sama dengan yang kami miliki dengan wadah IO kami.

Kode async tanpa pengecualian

Kita telah membahas cara kerja Result untuk kode murni dan tidak murni. Ide utamanya adalah: kami tidak memunculkan pengecualian, kami mengembalikannya. Hal ini sangat penting dalam kode async, karena satu pengecualian dapat merusak semua coroutine yang berjalan dalam satu eventloop.

Kami memiliki kombinasi wadah Future dan Result yang berguna: FutureResult . Sekali lagi, ini persis seperti IOResult , tetapi untuk kode async yang tidak murni. Gunakan saat Future Anda mungkin mengalami masalah: seperti permintaan HTTP atau operasi sistem file.

Anda dapat dengan mudah mengubah coroutine liar apa pun menjadi FutureResult yang tenang :

 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

Menggunakan FutureResult akan menjaga kode Anda aman dari pengecualian. Anda selalu dapat await atau mengeksekusi di dalam eventloop FutureResult apa pun untuk mendapatkan sinkronisasi instance IOResult agar berfungsi dengannya secara sinkron.

Komposisi asinkron yang lebih baik

Sebelumnya, Anda harus melakukan banyak await saat menulis kode 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 )

Beberapa orang setuju dengan hal itu, tetapi beberapa orang tidak menyukai gaya imperatif ini. Masalahnya adalah tidak ada pilihan.

Namun kini, Anda bisa melakukan hal yang sama dengan gaya fungsional! Dengan bantuan wadah Future dan 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 (...),
)

Atau bahkan sesuatu yang sangat mewah:

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