🎐 🚴 👧🏾 Apa Asynchrony Seharusnya Dengan Python ♐️ 💆🏿 🕺

Dalam beberapa tahun terakhir, asyncsemantik dan kata kunci pemrograman asinkron telah merasuki banyak bahasa pemrograman populer: JavaScript , Rust , C # , dan banyak lainnya . Tentu saja, Python juga memilikinya async/await, itu diperkenalkan dengan Python 3.5.

Pada artikel ini, saya ingin membahas masalah kode asinkron, berspekulasi tentang alternatif, dan mengusulkan pendekatan baru untuk mendukung aplikasi sinkron dan asinkron pada saat yang bersamaan.

Warna fungsi

Ketika fungsi asynchronous dimasukkan dalam bahasa pemrograman, itu pada dasarnya terbagi menjadi dua. Fungsi merah muncul (atau asinkron) dan beberapa fungsi tetap biru (sinkron).

Masalah utamanya adalah bahwa fungsi biru tidak dapat memanggil fungsi merah, tetapi fungsi merah berpotensi menyebabkan fungsi biru. Dalam Python, misalnya, ini sebagian benar: fungsi asinkron hanya dapat memanggil fungsi non-pemblokiran sinkron. Tetapi tidak mungkin untuk menentukan dari deskripsi apakah fungsi tersebut memblokir atau tidak. Python adalah bahasa scripting.

Pemisahan ini mengarah pada pembagian bahasa menjadi dua subset: sinkron dan asinkron. Python 3.5 dirilis lebih dari lima tahun yang lalu, tetapi asyncmasih belum didukung sebaik kemampuan sinkronisasi Python.

Anda dapat membaca lebih lanjut tentang warna fungsi di artikel bagus ini .

Kode duplikat

Warna fungsi yang berbeda berarti duplikasi kode dalam praktiknya.

Bayangkan Anda sedang mengembangkan alat CLI untuk mengambil ukuran halaman web dan Anda ingin mempertahankan cara sinkron dan asinkron untuk melakukannya. Misalnya, ini diperlukan jika Anda sedang menulis perpustakaan dan tidak tahu bagaimana kode Anda akan digunakan. Dan ini bukan hanya tentang pustaka PyPI, tetapi juga tentang pustaka kita sendiri dengan logika umum untuk berbagai layanan, ditulis, misalnya, di Django dan aiohttp. Meskipun, tentu saja, sebagian besar aplikasi independen ditulis hanya secara sinkron atau hanya secara asinkron.

Mari kita mulai dengan pseudocode sinkron:

def fetch_resource_size(url: str) -> int:
    response = client_get(url)
    return len(response.content)

Terlihat bagus. Sekarang mari kita lihat analog asinkron:

async def fetch_resource_size(url: str) -> int:
    response = await client_get(url)
    return len(response.content)

Secara umum, ini adalah kode yang sama, tetapi dengan penambahan kata asyncdan await. Dan saya tidak mengada-ada - bandingkan contoh kode dalam tutorial di httpx:

kode sinkron ;
kode asinkron .

Ada gambaran yang persis sama.

Abstraksi dan komposisi

Ternyata Anda perlu menulis ulang semua kode sinkron dan mengatur di sana-sini asyncdan awaitagar program menjadi asinkron.

Dua prinsip dapat membantu memecahkan masalah ini. Pertama, mari kita tulis ulang pseudocode imperatif menjadi fungsional. Ini akan memungkinkan Anda untuk melihat gambar dengan lebih jelas.

def fetch_resource_size(url: str) -> Abstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

Anda bertanya apa metode ini .map, apa fungsinya. Ini adalah bagaimana komposisi abstraksi kompleks dan fungsi murni terjadi dalam gaya fungsional. Ini memungkinkan Anda untuk membuat abstraksi baru dengan status baru dari yang sudah ada. Misalkan client_get(url)awalnya kembali Abstraction[Response], dan panggilan .map(lambda response: len(response.content))mengubah respons ke instance yang diperlukan Abstraction[int].

Menjadi jelas apa yang harus dilakukan selanjutnya. Perhatikan betapa mudahnya kami berpindah dari beberapa langkah independen ke pemanggilan fungsi sekuensial. Selain itu, kami mengubah jenis respons: sekarang fungsi mengembalikan beberapa abstraksi.

Mari tulis ulang kode untuk bekerja dengan versi asinkron:

def fetch_resource_size(url: str) -> AsyncAbstraction[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

Satu-satunya hal yang berbeda adalah tipe pengembalian - AsyncAbstraction. Kode lainnya sama persis. Anda tidak perlu lagi menggunakan kata kunci asyncdan await. awaittidak digunakan sama sekali ( demi semuanya ini dimulai ), dan tanpanya tidak ada gunanya async.

Hal terakhir adalah memutuskan klien mana yang kita butuhkan: asynchronous atau synchronous.

def fetch_resource_size(
    client_get: Callable[[str], AbstactionType[Response]],
    url: str,
) -> AbstactionType[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

client_getsekarang menjadi argumen tipe yang dapat dipanggil yang mengambil string URL sebagai input dan mengembalikan beberapa tipe di AbstractionTypeatas objek Response. AbstractionType- baik Abstractionatau AsyncAbstractiondari contoh sebelumnya.

Ketika kita lewat Abstraction, kode tersebut berjalan secara sinkron, ketika AsyncAbstraction- kode yang sama otomatis mulai berjalan secara asinkron.

IOResult dan FutureResult

Untungnya, dry-python/returnsabstraksi yang benar sudah tersedia .

Izinkan saya memperkenalkan Anda pada alat Python yang sepenuhnya aman, ramah mypy, kerangka kerja-agnostik, dan sepenuhnya. Ini memiliki abstraksi yang luar biasa, nyaman, luar biasa yang dapat digunakan dalam proyek apa pun.

Opsi sinkron

Pertama, mari tambahkan dependensi untuk mendapatkan contoh yang dapat direproduksi

pip install returns httpx anyio

Selanjutnya, mari ubah pseudocode menjadi kode Python yang berfungsi. Mari kita mulai dengan opsi sinkron.

from typing import Callable
 
import httpx
 
from returns.io import IOResultE, impure_safe
 
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

Butuh beberapa perubahan untuk mendapatkan kode yang berfungsi:

Penggunaan IOResultEadalah cara fungsional menangani kesalahan IO sinkron ( pengecualian tidak selalu berfungsi ). Jenis berdasarkan Resultmemungkinkan Anda untuk mensimulasikan pengecualian, tetapi dengan nilai yang terpisah Failure(). Pintu keluar yang berhasil kemudian dibungkus dalam sebuah tipe Success. Biasanya tidak ada yang peduli dengan pengecualian, tapi kami peduli.
Gunakan httpxyang dapat menangani permintaan sinkron dan asinkron.
Gunakan fungsi impure_safeuntuk mengonversi tipe pengembalian httpx.getmenjadi abstraksi IOResultE.

Opsi asinkron

Mari kita coba melakukan hal yang sama dalam kode asinkron.

from typing import Callable
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
 
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    return client_get(url).map(
        lambda response: len(response.content),
    )
 
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

Anda lihat: hasilnya persis sama, tetapi sekarang kodenya berjalan secara asinkron. Namun, bagian utamanya tidak berubah. Namun, Anda perlu memperhatikan hal-hal berikut:

Serentak IOResultEdiubah menjadi asynchronous FutureResultE, impure_safe- on future_safe. Ia bekerja sama, tetapi mengembalikan abstraksi yang berbeda: FutureResultE.
Digunakan AsyncClientdari httpx.
Nilai yang dihasilkan FutureResultperlu dijalankan karena fungsi merah tidak dapat memanggil dirinya sendiri.
Utilitas anyiodigunakan untuk menunjukkan bahwa pendekatan ini bekerja dengan perpustakaan asynchronous: asyncio, trio, curio.

Dua dalam satu

Saya akan menunjukkan cara menggabungkan versi sinkron dan asinkron dalam satu jenis API yang aman.

Tipe Jenis Lebih Tinggi dan kelas tipe untuk bekerja dengan IO belum dirilis (mereka akan muncul di 0.15.0), jadi saya akan mengilustrasikan seperti biasa @overload:

from typing import Callable, Union, overload
 
import anyio
import httpx
 
from returns.future import FutureResultE, future_safe
from returns.io import IOResultE, impure_safe
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], IOResultE[httpx.Response]],
    url: str,
) -> IOResultE[int]:
    """Sync case."""
 
@overload
def fetch_resource_size(
    client_get: Callable[[str], FutureResultE[httpx.Response]],
    url: str,
) -> FutureResultE[int]:
    """Async case."""
 
def fetch_resource_size(
    client_get: Union[
        Callable[[str], IOResultE[httpx.Response]],
        Callable[[str], FutureResultE[httpx.Response]],
    ],
    url: str,
) -> Union[IOResultE[int], FutureResultE[int]]:
    return client_get(url).map(
        lambda response: len(response.content),
    )

Kami menggunakan dekorator untuk @overloadmendeskripsikan data masukan apa yang diperbolehkan dan jenis nilai pengembalian yang akan diberikan. Anda @overloaddapat membaca lebih lanjut tentang dekorator di artikel saya yang lain .

Panggilan fungsi dengan klien sinkron atau asinkron terlihat seperti ini:

# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>
 
# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

Seperti yang Anda lihat, fetch_resource_sizedalam varian sinkron itu segera mengembalikan IOResultdan menjalankannya. Sedangkan pada versi asynchronous, diperlukan event-loop, seperti pada coroutine biasa. anyiodigunakan untuk menampilkan hasil.

Dalam mypykode ini tidak ada komentar:

» mypy async_and_sync.py
Success: no issues found in 1 source file

Mari kita lihat apa yang terjadi jika ada yang tidak beres.

---lambda response: len(response.content),
+++lambda response: response.content,

mypy dengan mudah menemukan kesalahan baru:

» mypy async_and_sync.py
async_and_sync.py:33: error: Argument 1 to "map" of "IOResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Argument 1 to "map" of "FutureResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"
async_and_sync.py:33: error: Incompatible return value type (got "bytes", expected "int")

Sleight of hand and no magic: menulis kode asynchronous dengan abstraksi yang benar hanya membutuhkan komposisi kuno yang bagus. Tetapi fakta bahwa kami mendapatkan API yang sama untuk tipe yang berbeda benar-benar hebat. Misalnya, ini memungkinkan Anda untuk mengabstraksi dari cara kerja permintaan HTTP: secara sinkron atau asinkron.

Semoga contoh ini menunjukkan betapa hebatnya program asynchronous. Dan jika Anda mencoba dry-python / return , Anda akan menemukan lebih banyak hal menarik. Di versi baru, kami telah membuat primitif yang diperlukan untuk bekerja dengan Tipe Kinded Tinggi dan semua antarmuka yang diperlukan. Kode di atas sekarang dapat ditulis ulang seperti ini:

from typing import Callable, TypeVar

import anyio
import httpx

from returns.future import future_safe
from returns.interfaces.specific.ioresult import IOResultLike2
from returns.io import impure_safe
from returns.primitives.hkt import Kind2, kinded

_IOKind = TypeVar('_IOKind', bound=IOResultLike2)

@kinded
def fetch_resource_size(
    client_get: Callable[[str], Kind2[_IOKind, httpx.Response, Exception]],
    url: str,
) -> Kind2[_IOKind, int, Exception]:
    return client_get(url).map(
        lambda response: len(response.content),
    )


# Sync:
print(fetch_resource_size(
    impure_safe(httpx.get),
    'https://sobolevn.me',
))
# => <IOResult: <Success: 27972>>

# Async:
page_size = fetch_resource_size(
    future_safe(httpx.AsyncClient().get),
    'https://sobolevn.me',
)
print(page_size)
print(anyio.run(page_size.awaitable))
# => <FutureResult: <coroutine object async_map at 0x10b17c320>>
# => <IOResult: <Success: 27972>>

Lihat cabang `master`, itu sudah berfungsi di sana.

Lebih banyak fitur dry-python

Berikut adalah beberapa fitur dry-python berguna lainnya yang paling saya banggakan.

Fungsi yang diketik partialdan@curry .

from returns.curry import curry, partial
 
def example(a: int, b: str) -> float:
    ...
 
reveal_type(partial(example, 1))
# note: Revealed type is 'def (b: builtins.str) -> builtins.float'
 
reveal_type(curry(example))
# note: Revealed type is 'Overload(def (a: builtins.int) -> def (b: builtins.str) -> builtins.float, def (a: builtins.int, b: builtins.str) -> builtins.float)'

Ini memungkinkan Anda untuk menggunakan @curry, misalnya, seperti ini:

@curry
def example(a: int, b: str) -> float:
    return float(a + len(b))
 
assert example(1, 'abc') == 4.0
assert example(1)('abc') == 4.0

Pipa fungsional dengan inferensi tipe.

Dengan menggunakan plugin mypy kustom, Anda dapat membuat pipeline fungsional yang menampilkan tipe.

from returns.pipeline import flow
assert flow(
    [1, 2, 3],
    lambda collection: max(collection),
    lambda max_number: -max_number,
) == -3

Biasanya dalam kode yang diketik sangat merepotkan untuk bekerja dengan lambda, karena argumen mereka selalu bertipe Any. Inferensi mypymemecahkan masalah ini.

Dengan bantuannya, kita sekarang tahu lambda collection: max(collection)tipe apa Callable[[List[int]], int], tapi lambda max_number: -max_numbersederhana Callable[[int], int]. Dalam flowdapat melewati sejumlah argumen, dan mereka akan bekerja dengan baik. Semua berkat pluginnya.

RequiresContextFutureResultuntuk injeksi ketergantungan fungsional yang diketik .

Abstraksi di atas FutureResult, yang telah kita bicarakan sebelumnya, dapat digunakan untuk secara eksplisit meneruskan dependensi ke program asinkron dalam gaya fungsional.

Rencana untuk masa depan

Sebelum kami akhirnya merilis versi 1.0, kami harus menyelesaikan beberapa tugas penting:

Menerapkan Jenis Jenis yang Lebih Tinggi atau emulasinya ( masalah ).
Tambahkan kelas jenis yang tepat untuk mengimplementasikan abstraksi yang diperlukan ( masalah ).
Mungkin mencoba kompiler mypyc, yang berpotensi memungkinkan program Python beranotasi yang diketik untuk dikompilasi ke biner. Kemudian kode c dry-python/returnsakan bekerja beberapa kali lebih cepat ( masalah ).
Jelajahi cara baru untuk menulis kode fungsional dengan Python, seperti "do-notation" .

kesimpulan

Masalah apa pun dapat diselesaikan dengan komposisi dan abstraksi. Pada artikel ini, kita melihat bagaimana menyelesaikan masalah warna fungsi dan menulis kode sederhana, mudah dibaca, dan fleksibel yang berfungsi. Dan lakukan pengecekan tipe.

Cobalah dry-python / return dan bergabunglah dengan Pekan Python Rusia : di konferensi tersebut, pengembang inti dry-python Pablo Aguilar akan mengadakan lokakarya tentang penggunaan dry-python untuk menulis logika bisnis.

Apa Asynchrony Seharusnya Dengan Python