這段時(shí)間寫微信小程序看見callback hell非常炸毛，而且確實(shí)我不熟悉js的各種衍生工具(async/await, promise)，想著python的coroutine是yield模改出來的，準(zhǔn)備自己也造一個(gè)輪子。由于生成器的原理就是把一段代碼掛起然后可以讓調(diào)用者控制是否繼續(xù)下去，這個(gè)行為和Monad非常像。Monad本身的原理極其簡(jiǎn)單(bind, ret)，關(guān)鍵在于Haskell、gluon或者Idris之類的語言里面可以有語法糖，防止一個(gè)embedding hell，于是我也想到要用yield來模擬這個(gè)過程。大致思路是a = yield m_a讓調(diào)用者知道這個(gè)是一個(gè)monad，然后通過generator的send來把bind的參數(shù)賦值給a，把a(bǔ)看作是后面的函數(shù)的參數(shù)即可，以避免不斷寫嵌套的匿名函數(shù)。因此我把js里的callback hell也變成了一個(gè)CPS來寫(Cont Monad)，當(dāng)然畢竟我比較懶不想學(xué)js/ts/es567等等新特性，所以只好自己造輪子了，大家要是在大的團(tuán)隊(duì)里就別這樣了，省的被打死。

我們知道Haskell里的do-notation：

main = dox <- someIOMonady <- anotherIOMonadotherIOMonads

其實(shí)只是如下結(jié)構(gòu)

main :: IO () main = someIOMonad >>= (x -> (anotherIOMonad >>= (y -> otherIOMonads)))

就只是把do-notation變成一堆monad的bind，由于monad里的內(nèi)容必須通過一個(gè)函數(shù)才能拿到，所以現(xiàn)在所有實(shí)現(xiàn)monad的方法里都是不停地嵌套才行，那使用monad的意義并不明確，monad本質(zhì)上在于用一個(gè)語法可以重載任意流程控制(也算是一種CPS？)來節(jié)省代碼，節(jié)約代碼就是節(jié)約生命，寫括號(hào)相當(dāng)于慢性自殺，所以得想辦法避開這個(gè)矛盾。

我們先簡(jiǎn)單實(shí)現(xiàn)一個(gè)Maybe，方便后面的說明：

class Maybe:def __init__(self, data, nothing=False):self.data = dataself._nothing = nothing@staticmethoddef nothing():return Maybe(None, nothing=True)@propertydef just(self):return not self._nothing@staticmethoddef ret(value):return Maybe(value, nothing=False)def bind(self, f):if self._nothing:return selfreturn f(self.data)def __repr__(self):if self.just:return f"Just {self.data}"return "Nothing"

以及一個(gè)簡(jiǎn)單的使用：

def my_input(prompt="input a number: ") -> Maybe:v = input(prompt)for x in v:if x not in "0123456789":return Maybe.nothing()return Maybe.ret(int(v))

考慮如下python代碼：

def do_process():a1 = yield my_input('number1: ') # anchor 1a2 = yield my_input('number2: ') # anchor 2return Maybe(a1 + a2)

假設(shè)有g(shù)en = do_process()是取得的generator
那么next(gen)將得到anchor 1位置的my_input所得到的結(jié)果， 對(duì)于剩下的那一部分來說，我們只要gen.send就可以把值發(fā)送給a1，剛好對(duì)應(yīng)了bind行為

因此考慮函數(shù)reduce，它的作用就是把一個(gè)不斷yield monad以及return monad的generator計(jì)算成最終return的結(jié)果

def reduce(generator):try:m_a = next(generator)except StopIteration as err:return err.valuereturn m_a.bind(lambda a: reduce(輸入a之后所形成的新的generator))

中間不難發(fā)現(xiàn)，我們需要一個(gè)輸入a，然后生成新的generator的函數(shù)來保持繼續(xù)reduce，我把這個(gè)過程稱為partial_apply:

def partial_apply(generator, x):while True:try:x = yield generator.send(x)except StopIteration as err:return err.value

實(shí)現(xiàn)也非常簡(jiǎn)單，但是我發(fā)現(xiàn)，這樣做相當(dāng)于每次都生成了一個(gè)新的generator，并且舊的generator還在那里，這是沒必要的，所以設(shè)計(jì)一個(gè)自己的Generator來輔助：

class _Generator:class _Null:passdef __init__(self, generator):self._generator = generatorself._x = _Generator._Null@staticmethoddef new(f):@wraps(f)def wrapper(*args, **kwargs):return _Generator(f(*args, **kwargs))return wrapperdef __next__(self):if self._x is not _Generator._Null:x = self._xself._x = _Generator._Nullreturn self._generator.send(x)return next(self._generator)def __iter__(self):return selfdef send(self, x):return self._generator.send(x)def partial_apply(self, x):self._x = xreturn selfdef __repr__(self):return f"Generator {self._generator.__repr__()}"

以及對(duì)應(yīng)的reduce函數(shù)

def reduce(generator: _Generator):try:m_a = generator.__next__()except StopIteration as err:return err.valuereturn m_a.bind(lambda a: reduce(generator.partial_apply(a)))

這里的遞歸我要說明一下，考慮Cont r a，由于continuation是需要bound f的運(yùn)算結(jié)果來繼續(xù)運(yùn)算的，所以沒辦法把它沒完全reduce的generator當(dāng)成這個(gè)f來用，而且這個(gè)generator只能用reduce來計(jì)算（否則沒辦法正確bind），所以這個(gè)遞歸我暫時(shí)想不到好方法來避免，以及我不知道怎么手寫棧來避開反復(fù)調(diào)用的開銷，當(dāng)然這個(gè)項(xiàng)目我放到GitHub上了，如果有誰有好想法可以發(fā)pull request

進(jìn)一步的，為了方便使用，我還加了一個(gè)decorator來輔助，思路都很簡(jiǎn)單，就不多做贅述了

from functools import wrapsdef do(generator_func):@wraps(generator_func)def wrapper(*args, **kwargs):return reduce(generator_func(*args, **kwargs))return wrapper

最終的使用效果是這樣的(來自我那個(gè)倉(cāng)里的實(shí)現(xiàn)):

from math import sqrtfrom functionalpy.monad import do, Eitherdef safe_div(a, b) -> Either:if b == 0:return Either.left("ZeroDivision")return Either.ret(a/b)def safe_sqrt(a) -> Either:if a < 0:return Either.left("Sqrt of a negative number")return Either.ret(sqrt(a))@do def cal(a, b):root = yield safe_sqrt(a)quotient = yield safe_div(root, b)return Either.ret(quotient)if __name__ == '__main__':print(cal(1, 2))print(cal(1, 0))print(cal(-1, 1))print(cal(-1, 0))

輸出

Right 0.5 Left ZeroDivision Left Sqrt of a negative number Left Sqrt of a negative number

雖然這個(gè)做法性能堪憂，但是對(duì)于js而言速度還是能接受的而且如果yield的次數(shù)不多的話，調(diào)用次數(shù)也不會(huì)太多，相比IO的等待以及callback hell，這些還是更容易接受的

然后是一個(gè)我在js里的實(shí)現(xiàn)（當(dāng)然我js寫的很爛了，而且很多的類型似乎因?yàn)槿鄙賖igher order type，也就是Haskell的kind或者idris的Type n，我沒辦法表達(dá)出來，所以到處都是any了，有誰能審計(jì)一下是最好的）

interface F<a, b> {(input: a): b }type Callback<a, r=void> = F<a, r> type Cont<r, a> = F<F<a, r>, r>class Pro<a> {cont: Cont<void, a>constructor(cont: Cont<void, a>) {this.cont = cont}run(callback: Callback<a, void>=console.log): void {this.cont(callback)}static ret<x>(value: x): Pro<x> {return new Pro(f => f(value))}bind<b>(f: F<a, Pro<b>>): Pro<b> {return new Pro(callback => {// this.callback: (a -> r) -> r// callback: (b -> r) -> r// f: a -> m b// return: rthis.run(function(x: a) {f(x).run(callback)})})} }interface Monad<a> {ret: (value: a) => Monad<a>bind: <b>(f: (input: a) => Monad<b>) => Monad<b> }interface Result<a=any> {done: booleanvalue: a }interface Gen<a=any> {next: (x?: any) => Result<a> }class _Null {}class _Gen<a=any> {gen: Genx: anyconstructor(gen: Gen) {this.gen = genthis.x = new _Null()}partialApply(x: any) {this.x = xreturn this}next(x?: any): Result<a> {if(this.x instanceof _Null) {return this.gen.next(x)}let v = this.xthis.x = new _Null()return this.gen.next(v)} }function reduce(gen: Gen): any {let _gen = new _Gen(gen)var result = _gen.next()if(result.done) {return result.value}var mA: Monad<any> = result.valuereturn mA.bind(a => reduce(_gen.partialApply(a))) }function monad(generatorFunc: any): any {return function(...args: any[]): any {return reduce(generatorFunc(...args))} }function wrap(f: (...args: any[]) => void): any {return function(...args: any[]): Pro<any> {return new Pro(callback => {f(...args, callback)})} }function coroutine<a>(generatorFunc: any): Pro<a> {return monad(generatorFunc) }function p<a>(generatorFunc: F<void, any>): Pro<a> {return monad(generatorFunc)() }function pret<a>(v: a): Pro<a> {return Pro.ret(v) }

使用例子是這樣的，當(dāng)然普通callback的調(diào)用也就不言而喻了

function callAsync(a: any, f: any) {return f(a)} let call = wrap(callAsync) p(function* () {let a = yield call(1)let b = yield call(2)return pret(a + b) }).run(console.log)

其實(shí)這個(gè)東西是為了做這就事的：
考慮如下結(jié)構(gòu)：

switch(status) {case null:weixinLoginwebLogingetUserInfobreakcase weixin:webLogingetUserInfobreakcase login:getUserInfobreak }

然而注意的switch可以省去break，于是有

switch(status) {case null:weixinLogincase weixin:webLogincase login:getUserInfo }

可以節(jié)約不少代碼的

但是如果js只能異步調(diào)用

switch(status) {case null:weixinLogin({success: () => {webLogin({success: () => {getUserInfo({success: callback})}})}})breakcase weixin:webLogin({success: ()=>{getUserInfo({success: callback})}})breakcase login:getUserInfo({success: callback})break }

這樣代碼量就相當(dāng)大了，所以說模改這個(gè)假裝是同步的代碼就很必要了

與50位技術(shù)專家面對(duì)面20年技術(shù)見證，附贈(zèng)技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的just函数 python_在python里写Monad的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。