在python使用类似于scala的链式编程方法

1.实现函数装饰器（随便copy了一个）

1.  
   T = TypeVar('T')
2.  
    
3.  
    
4.  
   class PyObject(ctypes.Structure):
5.  
   class PyType(ctypes.Structure):
6.  
   pass
7.  
    
8.  
   ssize = ctypes.c_int64 if ctypes.sizeof(
9.  
   ctypes.c_void_p) == 8 else ctypes.c_int32
10.  
    _fields_ = [
11.  
    ('ob_refcnt', ssize),
12.  
    ('ob_type', ctypes.POINTER(PyType)),
13.  
    ]
14.  
     
15.  
     
16.  
    def sign(clazz, funcName): # * 功能注册装饰器,加在函数上,可以将函数注册到特定类
17.  
    """
18.  
    ? clazz class类名
19.  
    ? funcName 注册的函数名称
20.  
    """
21.  
    def _(function):
22.  
    class SlotsProxy(PyObject):
23.  
    _fields_ = [('dict', ctypes.POINTER(PyObject))]
24.  
     
25.  
    name, target = clazz.__name__, clazz.__dict__
26.  
    proxy_dict = SlotsProxy.from_address(id(target))
27.  
    namespace = {}
28.  
    ctypes.pythonapi.PyDict_SetItem(
29.  
    ctypes.py_object(namespace),
30.  
    ctypes.py_object(name),
31.  
    proxy_dict.dict,
32.  
    )
33.  
    namespace[name][funcName] = function
34.  
     
35.  
    return _

2.注册函数方法

dict:

1.  
    
2.  
   # * dict
3.  
    
4.  
    
5.  
   @sign(dict, 'vs')
6.  
   def vs(self: dict) -> list:
7.  
   return list(self.values())
8.  
    
9.  
    
10.  
    @sign(dict, 'ks')
11.  
    def ks(self: dict) -> list:
12.  
    return list(self.keys())
13.  
     
14.  
     
15.  
    @sign(dict, 'kvs')
16.  
    def kvs(self: dict) -> list:
17.  
    re = []
18.  
    for k, v in self.items():
19.  
    re.append((k, v))
20.  
    return re
21.  
     
22.  
     
23.  
    @sign(dict, 'toStr')
24.  
    def dictToStr(self) -> str:
25.  
    return json.dumps(self, ensure_ascii=False)

list:

1.  
    
2.  
    
3.  
   @sign(list, 'map')
4.  
   def mapWith(self, __func: Callable[[Any], Any]):
5.  
   return list(map(__func, self))
6.  
    
7.  
   # * list
8.  
    
9.  
    
10.  
    @sign(list, 'flatMap')
11.  
    def flatMapWith(self, __func: Callable[[Any], List[Any]]):
12.  
    res = []
13.  
    for ts in map(__func, self):
14.  
    if ts:
15.  
    for t in ts:
16.  
    res.append(t)
17.  
    return res
18.  
     
19.  
     
20.  
    @sign(list, 'appendAll')
21.  
    def appendAll(self, vs: List[T]) -> List[T]:
22.  
    for v in vs:
23.  
    self.append(v)
24.  
    return self
25.  
     
26.  
     
27.  
    @sign(list, 'toSet')
28.  
    def toSet(self):
29.  
    return set(self)
30.  
     
31.  
     
32.  
    @sign(list, 'filter')
33.  
    def filterWith(self, __func: Callable[[Any], bool]):
34.  
    return list(filter(__func, self))
35.  
     
36.  
     
37.  
    @sign(list, 'distinct')
38.  
    def distinct(self):
39.  
    return list(set(self))
40.  
     
41.  
     
42.  
    @sign(list, 'groupByKey')
43.  
    def groupByKey(self: List[Tuple[str, T]]) -> Dict[str, List[T]]: # ? 聚合函数
44.  
    tmpMap = dict()
45.  
    for data in self:
46.  
    key = data[0]
47.  
    value = data[1]
48.  
    valueList = []
49.  
    if key in tmpMap:
50.  
    valueList = tmpMap[key]
51.  
    valueList.append(value)
52.  
    tmpMap[key] = valueList
53.  
     
54.  
    return tmpMap
55.  
     
56.  
     
57.  
    @sign(list, 'toDict')
58.  
    def tupleToDict(self: List[Tuple[str, T]]) -> Dict[str, T]:
59.  
    re = dict()
60.  
    for t in self:
61.  
    re[t[0]] = t[1]
62.  
     
63.  
    return re
64.  
     
65.  
     
66.  
    @sign(list, 'reduce')
67.  
    def reduceWith(self, __func):
68.  
    return reduce(__func, self)
69.  
     
70.  
     
71.  
    @sign(list, 'sum')
72.  
    def sumWith(self: List):
73.  
    return sum(self)
74.  
     
75.  
     
76.  
    @sign(list, 'reduceByKey')
77.  
    def reduceByKey(self: List[Tuple[str, T]], __func: Callable[[T], T]) -> Dict[str, T]:
78.  
    re = self.groupByKey()\
79.  
    .kvs()\
80.  
    .map(lambda t: (t[0], t[1].reduce(__func)))\
81.  
    .toDict()
82.  
    return re
83.  
     
84.  
     
85.  
    @sign(list, 'foreach')
86.  
    def foreach(self, __func):
87.  
    for t in self:
88.  
    __func(t)
89.  
    return self
90.  
     
91.  
     
92.  
    @sign(list, 'toStr')
93.  
    def listToStr(self) -> str:
94.  
    return json.dumps(self, ensure_ascii=False)
95.  
     
96.  
     
97.  
    @sign(list, 'isEmpty')
98.  
    def listIsEmpty(self: List) -> bool:
99.  
    return len(self) == 0
100.  
      

set:

1.  
   # * set
2.  
    
3.  
    
4.  
   @sign(set, 'toList')
5.  
   def toList(self):
6.  
   return list(self)

str:

1.  
    
2.  
   # * str
3.  
    
4.  
    
5.  
   @sign(str, 'startsIn')
6.  
   def startsIn(self: str, *keys):
7.  
   for s in keys:
8.  
   if self.startswith(s):
9.  
   return True
10.  
    return False
11.  
     
12.  
     
13.  
    @sign(str, 'endsIn')
14.  
    def endsIn(self: str, *keys):
15.  
    for s in keys:
16.  
    if self.endswith(s):
17.  
    return True
18.  
    return False
19.  
     
20.  
     
21.  
    @sign(str, 'append')
22.  
    def appendStr(self: str, s: str):
23.  
    return self str(s)

3.使用示例

1.  
    
2.  
   data=[("a",1),("b",2),("a",3),("b",4),("c",5)]
3.  
    
4.  
   re=data.map(lambda t: (t[0],t[1] 1))\
5.  
   .groupByKey()\
6.  
   .vs()\
7.  
   .map(lambda l:l.sum())\
8.  
   .reduce(lambda a,b:a b)
9.  
    
10.  
    print(re)
11.  
     
12.  
    """
13.  
    step 1: [("a",2),("b",3),("a",4),("b",5),("c",6)]
14.  
    step 2: {"a":[2,4],"b":[3,5],"c":[6]}
15.  
    step 3: [[2,4],[3,5],[6]]
16.  
    step 4: [6,8,6]
17.  
    step 5: 20
18.  
    """

这篇好文章是转载于：学新通技术网