Python并非經(jīng)典的FP(Functional Programming, 函數(shù)式編程)語(yǔ)言，用其原生的map/filter/reduce寫(xiě)法來(lái)實(shí)現(xiàn)函數(shù)式編程顯得相當(dāng)臃腫；好在可以通過(guò)第三方庫(kù)提供的語(yǔ)法糖來(lái)簡(jiǎn)化代碼。
本文將簡(jiǎn)明扼要地示范兩個(gè)常用庫(kù)的寫(xiě)法；讀完，你就能寫(xiě)出一手優(yōu)雅而簡(jiǎn)潔的函數(shù)式Python代碼了

• 使用PyFunctional來(lái)提供類(lèi)似Streaming API，形成鏈?zhǔn)秸{(diào)用
• 使用fn.py來(lái)簡(jiǎn)化lambda表達(dá)式的書(shū)寫(xiě)

背景

FP (Functional Programming, 函數(shù)式編程) 是一種有效、簡(jiǎn)潔、優(yōu)雅，且對(duì)并發(fā)相當(dāng)友好的編程范式。
可惜python并不原生地支持FP；原因之一是Python的作者Guido大爺并非FP的粉絲，下面是老爺子對(duì)FP的看法^[1]

I have never considered Python to be heavily influenced by functional languages, no matter what people say or think. I was much more familiar with imperative languages such as C and Algol 68 and although I had made functions first-class objects, I didn't view Python as a functional programming language. However, earlier on, it was clear that users wanted to do much more with lists and functions.

...

It is also worth noting that even though I didn't envision Python as a functional language, the introduction of closures has been useful in the development of many other advanced programming features. For example, certain aspects of new-style classes, decorators, and other modern features rely upon this capability.

Lastly, even though a number of functional programming features have been introduced over the years, Python still lacks certain features found in “real” functional programming languages. For instance, Python does not perform certain kinds of optimizations (e.g., tail recursion). In general, because Python's extremely dynamic nature, it is impossible to do the kind of compile-time optimization known from functional languages like Haskell or ML. And that's fine.

--- Guido van Rossum

經(jīng)過(guò)一段時(shí)間的摸索和總結(jié)，我找到了一套適用于Python的比較簡(jiǎn)潔的FP最佳實(shí)踐。具體來(lái)說(shuō)就是使用文章開(kāi)頭提到的兩個(gè)庫(kù)，下面分別演示一下這兩個(gè)庫(kù)的基礎(chǔ)用法，并結(jié)合幾個(gè)實(shí)際工作中的例子略作演示。
注意：本文既然是寫(xiě)B(tài)CP，就打算只講干貨，盡量少些奇技淫巧；如果想進(jìn)一步深入了解；我會(huì)在文末給出一些閱讀材料

PyFunctional

PyFunctional提供了Streaming API，非常方便序列的鏈?zhǔn)讲僮鳎送膺€有提供一些IO小工具。
下面來(lái)看幾個(gè)例子。

IO

pip install pyfunctional

%%bash
mkdir ./res
cat << EOF > ./res/json_lines.json
{"name":"Alice", "age":5}
{"name":"Bob", "age":6}
EOF

from functional import seq # 注意包名不是pyfunctional
seq.jsonl('./res/json_lines.json')
# [{'age': 5, 'name': 'Alice'}, {'age': 6, 'name': 'Bob'}]

seq.jsonl('./res/json_lines.json').to_pandas()

to_pandas.png

Streaming API

# range 的用法
assert seq.range(5) == seq(range(5)) 
assert seq.range(5) == seq([0,1,2,3,4])
assert seq.range(5) == seq(0,1,2,3,4)
assert seq.range(5) == [0,1,2,3,4]

accumulate / aggregate

accumulate是aggregate的一個(gè)特例；所以已經(jīng)deprecated
aggregate支持1～3個(gè)參數(shù)，分別代表

• arg1: init_value=None， 聚合開(kāi)始時(shí)所用的初始值
• arg2: fn, (current, next) ==> result， 兩兩聚合所用的函數(shù)
• arg3: agg_func=None， 在返回結(jié)果前執(zhí)行的最后一個(gè)映射

seq.range(5).aggregate(-2, operator.add, str)
# '8'

Cartesian 笛卡爾積

s1 = range(2)
s2 = set('abc')
seq(s1).cartesian(s2).to_list() # 求s1與s2的笛卡爾積
# [(0, 'a'), (0, 'b'), (0, 'c'), (1, 'a'), (1, 'b'), (1, 'c')]

count / count_by_key / count_by_value

f = X%2==0
seq.range(5).count(f) # 這個(gè)f必填
# 3

# count_by_key 已經(jīng)過(guò)時(shí)，用 reduce_by_key近似
seq([('a', 1), ('b', 2), ('b', 3), ('b', 4), ('c', 3), ('c', 0)]) .reduce_by_key(operator.add).to_list()
# [('a', 1), ('b', 9), ('c', 3)]

# count_by_value不可用，用Counter近似
from collections import Counter
s = seq(['a', 'a', 'a', 'b', 'b', 'c', 'd'])
Counter(s)
# Counter({'a': 3, 'b': 2, 'c': 1, 'd': 1})

dict

import numpy as np

# 可以直接傳固定值
# d = seq([('a', 1), ('b', 2)]).dict('nan')

# 也可以傳default_func，就像 defaultdict(default_func) 一樣
d = seq([('a', 1), ('b', 2)]).dict(np.random.rand)
d['a'],d['c'],d['d'],d['e'] # 每次求新值的時(shí)候，要運(yùn)行一下defaultdict的init_value函數(shù)
# (1, 0.1045269208888322, 0.2721328917391167, 0.8890019300218747)

d['a'],d['c'],d['d'],d['e'] # 多次執(zhí)行有緩存
# (1, 0.1045269208888322, 0.2721328917391167, 0.8890019300218747)

drop / drop_right / drop_while

seq([1, 2, 3, 4, 5]).drop(2) # 去掉開(kāi)頭的2個(gè)元素
seq([1, 2, 3, 4, 5]).drop_right(2) # 去掉結(jié)尾的2個(gè)元素
seq([1, 2, 3, 4, 5, 9, 2]).drop_while(X < 3) # 一直drop，直到遇到條件為False
# [3, 4, 5, 9, 2]

exists / for_all

seq(1,2).exists(X>=2)
# True

def is_even_log(n):
    print('log: {}'.format(n))
    return n%2==0
seq.range(10).for_all(is_even_log) # for_all是當(dāng)且僅當(dāng)序列中的全部元素都能算出True時(shí)，才為T(mén)rue；從log看，有l(wèi)azy_eval
# log: 0
# log: 1
# Out[31]: False

flatmap / flatten

seq([[1, 2], [3, 4], [5, 6]]).flatten() # 將 arr_or_arr 展開(kāi)打平 
# [1, 2, 3, 4, 5, 6]

arr_of_arr = [[1, 2, 3], [3, 4], [5, 6]]
fn = X
seq(arr_of_arr).flat_map(lambda a: [min(a),]*4)
# 對(duì)arr_of_arr中的每個(gè)子序列執(zhí)行fn映射，得到新的子序列（記作ARR）；然后對(duì)ARR組成的大序列執(zhí)行flatten
# [1, 1, 1, 1, 3, 3, 3, 3, 5, 5, 5, 5]

fold_left / fold_right

# 從起始值開(kāi)始， 逐個(gè)調(diào)用傳入的fold函數(shù)
seq.range(3).fold_left('a', lambda c,n: '{}_{}'.format(c,n) ) 
# 'a_0_1_2'

# 從起始值開(kāi)始， 逐個(gè)調(diào)用傳入的fold函數(shù)； 注意lambda中兩個(gè)參數(shù)的順序互換了
seq.range(3).fold_right('a', lambda n,c: '{}_{}'.format(c,n) ) 
# 'a_2_1_0'

group_by / group_by_key / grouped

seq(["abc", "ab", "z", "f", "qw"]).group_by(len).list()
# [(1, ['z', 'f']), (2, ['ab', 'qw']), (3, ['abc'])]

seq([('a', 1), ('b', 2), ('b', 3), ('b', 4), ('c', 3), ('c', 0)]).group_by_key().list()
# [('a', [1]), ('b', [2, 3, 4]), ('c', [3, 0])]

# group支持相鄰元素大致分組，可以用slicing近似（后文有提到）
# 這里的兩個(gè)list函數(shù)，是為了方便觀(guān)察（不然就打印出一堆generator）
seq([1, 2, 3, 4, 5, 6, 7, 8]).grouped(3).map(list).list()
# [[1, 2, 3], [4, 5, 6], [7, 8]]

init / inits / tail / tails

assert seq.range(5).init() == seq.range(5).drop_right(1)
# 命名很詭異，init不是(initialization, 初始化)，而是除最后一個(gè)元素以外的子序列
seq.range(5).init()
# [0, 1, 2, 3]

seq.range(5).inits().list() # inits 是含開(kāi)頭元素的所有子序列
# [[0, 1, 2, 3, 4], [0, 1, 2, 3], [0, 1, 2], [0, 1], [0], []]

seq.range(5).tail() # 除開(kāi)頭元素以外的子序列
# [1, 2, 3, 4]

seq.range(5).tails().list() # 含尾元素在內(nèi)的所有子序列
# [[0, 1, 2, 3, 4], [1, 2, 3, 4], [2, 3, 4], [3, 4], [4], []]

join

seq([('a', 1), ('b', 2), ('c', 3)]).join([('a', 2), ('c', 5)], "inner") # inner是默認(rèn)行為，可省略
# [('a', (1, 2)), ('c', (3, 5))]

seq([('a', 1), ('b', 2)]).join([('a', 3), ('c', 4)], "left")
# [('a', (1, 3)), ('b', (2, None))]

seq([('a', 1), ('b', 2)]).join([('a', 3), ('c', 4)], "right")
seq([('a', 1), ('b', 2)]).join([('a', 3), ('c', 4)], "outer")

make_string

其實(shí)就是str.join，但是join關(guān)鍵字已經(jīng)被用了

seq(['a','b',1,{'name':'jack'}]).make_string('@')
# "a@b@1@{'name': 'jack'}"

order_by

from fn import F # 這是fn.py提供的復(fù)合函數(shù)語(yǔ)法糖，后文有講
seq(1,'abc',55,9999,718).order_by(F(str)>>len)
# [1, 55, 'abc', 718, 9999]

partition

seq.range(-5,5).partition(X>0) # 返回 (truthy, falsy)
# [[1, 2, 3, 4], [-5, -4, -3, -2, -1, 0]]

reduce_by_key

seq([('a', 1), ('b', 2), ('b', 3), ('b', 4), ('c', 3), ('c', 0)]).reduce_by_key(X+X)
# [('a', 1), ('b', 9), ('c', 3)]

sliding 滑動(dòng)窗口

第一個(gè)參數(shù)是size，第二個(gè)是step(默認(rèn)=1)

# 模擬100ms的語(yǔ)音，每25ms為一幀，滑動(dòng)窗口為10ms
seq.range(100).sliding(25, 10)

# [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24],
# [10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34],
# ...

sorted

seq('a1','c9','b3','b2').sorted(key=X[1], reverse=True)
# ['c9', 'b3', 'b2', 'a1']

starmap / smap

starmap是把序列中的元素作為參數(shù)，逐個(gè)用func作映射

seq([(2, 3), (-2, 1), (0, 10)]).smap(X+X)
# [5, -1, 10]

zip / zip_with_index

seq.range(3).zip('abc').list()
# [(0, 'a'), (1, 'b'), (2, 'c')]

seq(list('abc')).zip_with_index(start=9).list()
# [('a', 9), ('b', 10), ('c', 11)]

fn.py

fn.py提供了一些高級(jí)的FP工具，本身也提供了類(lèi)似于PyFunctional的一部分基礎(chǔ)FP功能(e.g. map/reduce/filter)；但是這些功能在fn.py中不支持鏈?zhǔn)秸{(diào)用語(yǔ)法，不太方便，本文就不講了
除此之外，fn.py還提供了一些高級(jí)的FP特性，包括：

• underscore表達(dá)式
• 函數(shù)復(fù)合管道F()
• 無(wú)窮Stream
• Monad Optional
• TCO 尾遞歸優(yōu)化
• Immutable容器
• ...

初次看到這些概念的同學(xué)可以會(huì)對(duì)一部分術(shù)語(yǔ)感到陌生，但是不用擔(dān)心，后面我會(huì)在例子中講清楚

underscore表達(dá)式

熟悉前端的同學(xué)應(yīng)該聽(tīng)說(shuō)過(guò)，有個(gè)Javascript庫(kù)叫作Underscore.js；它提供了一種用于書(shū)寫(xiě)lambda表達(dá)式的語(yǔ)法糖^[2]
fn.py也提供了類(lèi)似的功能

Identity 常函數(shù)

先安裝： pip install fn

from fn import _ as X # 在ipython環(huán)境中，下劃線(xiàn)有特征語(yǔ)義，要回避開(kāi)
f = X # 空的underscore本身就是Identity函數(shù)
print(f) # (x1) => x1
print(f(10)) # 10

Arithmetic 基礎(chǔ)算術(shù)

f = X+2
f(3) # 5

# 還支持 乘方/取模/移位/取反/位運(yùn)算 ...
(X ** 2)(3) # 9
(X % 2)(3) # 1
(X << 2)(8) # 32
(1 & X)(0) # 0

多參數(shù)

(X-X)(5,3) # 2

property/attribute getter

class Foo(object):
    def __init__(self):
        self.attr1 = 10
        self.attr2 = 'mystr'
    @property
    def prop1(self):
        return self.attr2.upper()
    def method1(self):
        return self.attr2.capitalize()
    
foo = Foo()
(X.attr1*-2)(foo) # -20
(X.attr2)(foo) # 'mystr'
(X.prop1)(foo) # 'MYSTR'
(X.method1())(foo) # ArityError, 不支持這樣調(diào)method，正確用法見(jiàn)下文

method caller

f = X.call('method1')
f(foo) # 'Mystr'

f = X.call('split','-')
f('abc-def') # ['abc', 'def']

d = {'num':32}
f = X.call('update', num=42)
f(d) # None
d # {'num': 42}

str

可以用一種可能讀較好的方式把underscore表達(dá)式打印出來(lái)，還能自動(dòng)處理優(yōu)先級(jí)

f = X + X * X 
str(f) # '(x1, x2, x3) => (x1 + (x2 * x3))'

函數(shù)復(fù)合管道F()

from fn import F

f = X * 2
g = X + 10
h = X * 10

func = F(f) << g << h
func(5) # [(?*10) + 10] * 2
# 注意 << 是向內(nèi)復(fù)合，好記：箭頭從右向左指，表示先執(zhí)行右邊的，再代入左邊的函數(shù)

(F(f) << F(g) << F(h))(5)  # 多個(gè)F對(duì)象也可以復(fù)合

# F()中預(yù)填入一部分參數(shù)，就是partial
import operator
f = F(operator.sub, 10)
f(3) # 7

# Pipe Composition 向外復(fù)合
f = F(X**2) >> (X + 10)
f(5) # 35

無(wú)窮Stream

這是一個(gè)比較難懂的語(yǔ)法，我也沒(méi)空去讀源碼搞透徹；大概意思就是一個(gè)懶漢式的生成器的串聯(lián)。

Lazy-evaluated Scala-style streams. Basic idea: evaluate each new element "on demand" and share calculated elements between all created iterators. Stream object supports << operator that means pushing new elements when it's necessary.

--- 官網(wǎng)解釋

下面的demo盡量體會(huì)一下就好，不求甚解。

from fn import Stream

# 遞歸slice
s = Stream() << range(10)
list(s[2:9][:3])  # s[2:9]是[2,3,4,5,6,7,8]，對(duì)這個(gè)Stream再取[:3]就是[2,3,4]

# fibonacci
s = Stream() << [0, 1]
fib = s << iters.map(operator.add, s, iters.drop(1, s))
# 注意最右邊的<<是一直在往Stream里面插入新值，導(dǎo)致s的cursor往后走
list(iters.take(5, fib))

Monad Optional^[3]

跟Java8的Optional差不多，用來(lái)包裹Nullable對(duì)象

基本用法

from fn import monad

monad.Option(10, checker=X > 70) # Empty()
monad.Empty(5) # Empty()
monad.Full(10) # Full(10)
monad.Option.from_value(10) # Full(10)
monad.Option.from_call(X**2, 5) # Full(25)
monad.Option.from_call(X[X], dict(k='v'), 'bad_k', exc=KeyError) # Empty()

# 自動(dòng)flatten，說(shuō)明這個(gè)庫(kù)內(nèi)部還是有優(yōu)化的
monad.Option(monad.Option(monad.Option(10))) # Full(10)

map/filter, 這個(gè)比較實(shí)用，用函數(shù)式的方法從Nullable中安全地取值

class Request(dict):  # 注意這里有個(gè)繼承
    def param(self, name):
        return monad.Option(self.get(name, None))  # 注意這里，手動(dòng)包了一層monad.Option

    # 注意，這個(gè)函數(shù)跟上面功能相同；只是自動(dòng)被包裹了一層 Option；不用自己寫(xiě)monad.Option(...)；可讀性更好，且對(duì)舊代碼改動(dòng)小
    @monad.optionable
    def optParam(self, name):
        return self.get(name, None)
# 定義好對(duì)象
req = Request(testing='Fixed', empty='')

# demo1: 順利取值
(req.param('testing')  # 拿到monad參數(shù)
 .map(operator.methodcaller('strip'))  # 去掉首尾空白字符
 .filter(len)  # 去掉空串
 .map(operator.methodcaller('upper'))  # 轉(zhuǎn)大寫(xiě)
 .get_or('empty_value')  # 取出值來(lái)
)
# 'FIXED'

# demo2: 注解式寫(xiě)法的optParam跟上面等效
(req.optParam('testing')  # optParam函數(shù)用的是注解式寫(xiě)法
 .map(operator.methodcaller('strip')).filter(len)
 .map(operator.methodcaller('upper')).get_or('empty_value'))
# 'FIXED'

# demo3: 用filter濾掉指定的值后，變成Empty
(req.param('empty').map(operator.methodcaller('strip'))
 .filter(len)  # 去掉空串, 變成Monad(None)
 .map(operator.methodcaller('upper')).get_or('empty_value'))
# 'empty_value'

# demo4: 一開(kāi)始就到Optional(None)，即Empty；后面的鏈?zhǔn)絤ap/filter步驟skip
(req.param('missing')  # missing字段不存在，拿到的是Monad.Empty
 .map(operator.methodcaller('strip')).filter(len)
 .map(operator.methodcaller('upper')).get_or('empty_value'))
# 'empty_value'

or_call調(diào)用，通過(guò)函數(shù)來(lái)取fallback值

# demo1: 直接取到值
r = dict(type='jpeg')
# 注意第一步就用monad.Option給包裹起來(lái)了，后面才能方便的鏈?zhǔn)秸{(diào)用or_call/map等方法
(monad.Option(r.get('type', None))  # 直接得到Full值，下面的or_call被跳過(guò)
 .or_call(from_mimetype, r)
 .or_call(from_extension, r)
 .map(operator.methodcaller('upper'))
 .get_or('unknown'))
# 'JPEG'
 
# demo2: 直接用返回值None表示Empty
from_mimetype = X.call('get', 'mimetype', None) # 可以直接返回值（None）表示Empty
r = dict(mimetype='png')
(monad.Option(r.get('type', None))  # monad.Empty
 .or_call(from_mimetype, r) # 上面是Emtpy，觸發(fā)這一步or_call的執(zhí)行,得monad.Full('png')
 .or_call(from_extension, r) # 已經(jīng)是Full值，這個(gè)or_call被跳過(guò)
 .map(operator.methodcaller('upper'))
 .get_or('unknown'))
 
# demo3: 顯式地返回Empty
def from_extension(r):
    return (monad.Option(r.get('url', None)).map(X.call('split', '.')[-1]))
r = dict(url='image.gif')
(monad.Option(r.get('type', None))  # monad.Empty
 .or_call(from_mimetype, r) # 這一步調(diào)用返回None,還是Empty
 .or_call(from_extension, r)  # 這個(gè)函數(shù)是顯式地返回了Empty，跟上面的demo等效,得monad.Full('gif')
 .map(operator.methodcaller('upper'))
 .get_or('unknown'))
 
# demo4: 最后的fallback
r = dict()
(monad.Option(r.get('type', None)) # 這次r是空字典，get到的是Option(None)，即Empty
 .or_call(from_mimetype, r) # 沒(méi)有mimetype信息，返回的是None，仍然Empty
 .or_call(from_extension, r)  # 同上，仍然Emtpy
 .map(operator.methodcaller('upper')) # Empty對(duì)象跳過(guò)map映射
 .get_or('unknown')) # 取到fallback值

TCO(Tail Call Optimization, 尾遞歸優(yōu)化)

對(duì)很多問(wèn)題來(lái)說(shuō)，遞歸是一種很直觀(guān)/順手的寫(xiě)法，描述能力強(qiáng)，不費(fèi)腦細(xì)胞。
然而，遞歸由于需要額外的函數(shù)調(diào)用開(kāi)銷(xiāo)，性能一般比循環(huán)的寫(xiě)法差一些；而且函數(shù)棧的嘗試是有限的，超過(guò)之后就會(huì)報(bào)錯(cuò)。
在某些語(yǔ)言中，內(nèi)置了尾調(diào)用優(yōu)化（建議讀一讀）的特性：當(dāng)最后一個(gè)語(yǔ)句是純粹的遞歸調(diào)用時(shí)，棧深度不變，只改動(dòng)參數(shù)表，性能上相當(dāng)于goto語(yǔ)句。
Python并非經(jīng)典的函數(shù)式語(yǔ)言，并沒(méi)有內(nèi)置TCO機(jī)制；但是可以用fn.py來(lái)近似。

準(zhǔn)備工作

!pip install memory_profiler
%load_ext memory_profiler
import sys
sys.getrecursionlimit() # 1000, 下面的python原生寫(xiě)法中，如果超過(guò)這個(gè)深度會(huì)報(bào)錯(cuò)

用經(jīng)典的遞歸寫(xiě)法實(shí)現(xiàn)fibonacci，測(cè)時(shí)間&內(nèi)存開(kāi)銷(xiāo)

%%time
def factorial_normal_recurive(n): # 最直觀(guān)的寫(xiě)法， f(n) = f(n-1) * n
    if n==1:
        return 1
    else:
        return factorial_normal_recurive(n-1) * n

%memit factorial_normal_recurive(900)
# peak memory: 38.27 MiB, increment: 0.86 MiB
# CPU times: user 46 ms, sys: 56.5 ms, total: 103 ms
# Wall time: 242 ms

寫(xiě)成尾遞歸，執(zhí)行時(shí)間快一點(diǎn)

%%time
def factorial_tail_recursive(n, acc=1): # 尾遞歸，把子遞歸的結(jié)果直接返回； f(n,acc) = f(n-1, acc*n)
    if n==1:
        return acc
    else:
        return factorial_tail_recursive(n-1, acc*n)
%memit factorial_tail_recursive(900)
# peak memory: 39.48 MiB, increment: 1.18 MiB
# CPU times: user 41.6 ms, sys: 30.4 ms, total: 72 ms
# Wall time: 208 ms

開(kāi)啟fn.py的TCO，調(diào)用棧深度可以超過(guò)解釋器的限制；而且會(huì)被翻譯成循環(huán)/goto調(diào)用，性能也更好（這個(gè)沒(méi)測(cè)）

# 為了方便TCO, 原先函數(shù)的返回語(yǔ)句要寫(xiě)成以下3種格式之一：
# (False, result) means that we finished
# (True, args, kwargs) means that we need to call function again with other arguments
# (func, args, kwargs) to switch function to be executed inside while loop # f1和f2相互遞歸調(diào)用時(shí)用到

%%time
from fn import recur
@recur.tco
def factorial_tco(n, acc=1): # 跟尾遞歸同理，但是寫(xiě)成了fn.py指定的格式；便于開(kāi)啟TCO自動(dòng)優(yōu)化
    if n==1:
        return False, acc
    else:
        return True, (n-1, acc*n)
%memit factorial_tco(90000) # 注意，這里的尾調(diào)用已經(jīng)被優(yōu)化成了循環(huán)，所以可以加大深度超過(guò)解釋器的限制
# 性能不用看，這次測(cè)試的是90000，比上面幾次實(shí)驗(yàn)大多了
# peak memory: 44.66 MiB, increment: 5.17 MiB
# CPU times: user 3.11 s, sys: 36.4 ms, total: 3.15 s
# Wall time: 3.3 s

Immutable容器

眾所周知，python中的str是immutable的，字符串無(wú)法inplace modify，只能通過(guò)賦值指向新的地址。^[4]

strA='hello'
print(id(strA)) # 4367427544
strA+=' world'
print(id(strA)) # 4373528112
print(strA) # hello world

這種immutable的設(shè)計(jì)，正好契合了FP中“無(wú)副作用的函數(shù)”的理念。

SkewHeap / PairingHeap

基礎(chǔ)用法

from fn.immutable import SkewHeap, PairingHeap, LinkedList, Stack, Queue, Vector, Deque

heap = SkewHeap()
for i in [9,3,5,2]:
    heap = heap.insert(i) # 注意是Immutable，不能inplace改，只能賦值改引用
list(heap) # 不指定cmp時(shí)，用的是自然排序
# [2, 3, 5, 9]

ele, remain = heap.extract()
ele, list(remain)
# (2, [3, 5, 9])

指定key/cmp

from collections import namedtuple
Student = namedtuple('Student', ['age','name'])
s1 = Student(7, 'Tora')
s2 = Student(5, 'Ponyo')
s3 = Student(6, 'Sousuke')

heap = SkewHeap(key=operator.attrgetter('age')) # 指定key
for s in [s1,s2,s3]:
    heap = heap.insert(s)
list(heap)
# [Student(age=5, name='Ponyo'),
#  Student(age=6, name='Sousuke'),
#  Student(age=7, name='Tora')]

heap = SkewHeap(cmp=X.age-X.age) # 指定cmp
for s in [s1,s2,s3]:
    heap = heap.insert(s)
list(heap)
# [Student(age=5, name='Ponyo'),
#  Student(age=6, name='Sousuke'),
#  Student(age=7, name='Tora')]

支持存儲(chǔ)Pair的堆

ph = PairingHeap() # 支持存Pair的堆
ph = ph.insert((15,'a'))
ph = ph.insert((12,'b'))
ph = ph.insert((13,'c'))
list(ph)
# [(12, 'b'), (13, 'c'), (15, 'a')]

Else

除了堆以外，還對(duì)常見(jiàn)的各種數(shù)據(jù)結(jié)構(gòu)進(jìn)行了Immutable封裝；不多說(shuō)了，詳見(jiàn)官網(wǎng)文檔

來(lái)幾枚大栗子

朕累了，寫(xiě)不動(dòng)了；以后遇到合適的例子就補(bǔ)到這里。

tired_meow.jpeg

More

感興趣的同學(xué)可以自己探索以下資源

• 一個(gè)輕量級(jí)的FP庫(kù) Streamz
• PyFunctional文檔
• 我用Jupyter Notebook寫(xiě)的fn.py/PyFunctionalDemo

1. 摘自StackOverflow上的一條回答 ?

2. ES6中的lambda表達(dá)式可能會(huì)好寫(xiě)一點(diǎn)，我了解不多，歡迎感興趣的同學(xué)評(píng)論留言 ?

3. Monad和Optional不是一碼事；前于前者可以看阮一峰的這篇文章，后者可以參考Java8 Optional來(lái)理解 ?

4. 關(guān)于python中的可變/不可變類(lèi)型，這篇文章總結(jié)得不錯(cuò) ?