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

我們知道Haskell里的do-notation：

main = dox <- someIOMonady <- anotherIOMonadotherIOMonads

其實只是如下結構

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

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

我們先簡單實現一個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"

以及一個簡單的使用：

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)

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

因此考慮函數reduce，它的作用就是把一個不斷yield monad以及return monad的generator計算成最終return的結果

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

中間不難發現，我們需要一個輸入a，然后生成新的generator的函數來保持繼續reduce，我把這個過程稱為partial_apply:

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

實現也非常簡單，但是我發現，這樣做相當于每次都生成了一個新的generator，并且舊的generator還在那里，這是沒必要的，所以設計一個自己的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__()}"

以及對應的reduce函數

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的運算結果來繼續運算的，所以沒辦法把它沒完全reduce的generator當成這個f來用，而且這個generator只能用reduce來計算（否則沒辦法正確bind），所以這個遞歸我暫時想不到好方法來避免，以及我不知道怎么手寫棧來避開反復調用的開銷，當然這個項目我放到GitHub上了，如果有誰有好想法可以發pull request

進一步的，為了方便使用，我還加了一個decorator來輔助，思路都很簡單，就不多做贅述了

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

最終的使用效果是這樣的(來自我那個倉里的實現):

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

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

然后是一個我在js里的實現（當然我js寫的很爛了，而且很多的類型似乎因為缺少higher order type，也就是Haskell的kind或者idris的Type n，我沒辦法表達出來，所以到處都是any了，有誰能審計一下是最好的）

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

使用例子是這樣的，當然普通callback的調用也就不言而喻了

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)

其實這個東西是為了做這就事的：
考慮如下結構：

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

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

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

可以節約不少代碼的

但是如果js只能異步調用

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

這樣代碼量就相當大了，所以說模改這個假裝是同步的代碼就很必要了

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。