Functional core as a pipeline in Python

The main goal of this post is to show one architecture pattern of little use in Python called " functional core - imperative wrapper ", in which functional programming is mixed with imperative programming in an attempt to negate the shortcomings of each of them. Functional languages ​​are known to be weak when interacting with the "real world", such as user input, GUI interaction, or other I / O operations.





This approach uses Python's ability to work with functions within the functional paradigm, in which functions can be manipulated in the same way as any other object: passed as arguments to other functions, returned from functions, and included in sequences as their elements.





For those looking to brush up on functional programming in Python, I recommend following the link to my post on the basics of FP in Python.





Data processing pipeline
Data processing pipeline

The functional style of programming is very close to how a person thinks while solving a problem. β€œLet it be given x



. In order to solve the problem with this data, it is necessary to perform a series of transformations. First, apply to them f



and get the resulting data x'



. Then apply to the new data f2



and get new resulting data x''



, etc.





, . , .. . , . , (), .





, , (1) (2) , debug, (3) , .





, . F#:





 2                            
 |> ( fun x -> x + 5)         
 |> ( fun x -> x * x)         
 |> ( fun x -> x.ToString() ) 
      
      



, 2, -. Python, , , , :





#   
 def pipe(data, *fseq):
    for fn in fseq: 
        data = fn(data)
    return data
      
      



Python:





pipe(2,
     lambda x: x + 5,
     lambda x: x * x,
     lambda x: str(x))
      
      



:





add      = lambda x: lambda y: x + y
square   = lambda x: x * x
tostring = lambda x: str(x)

pipe(2,
     add(5),
     square,
     tostring)
      
      



2 , '49'



. reduce



, , pipe .





pipe



: data



fseq



. for



. , data . .. , . pipe . .





. pipe *



. *



.





, , . ,





def my_sum(*args):  #   
    return sum(args)

my_sum(1, 2, 3, 4, 5)
      
      



. ,





def fun(a, b, c, d):
    print(a, b, c, d)

my_list = [1, 2, 3, 4]
fun(*my_list) #    
      
      



- ( ):





class Factory:
    def process(self, input):
        raise NotImplementedError

class Extract(Factory):
    def process(self, input):
        print(" ...")
        output = {}
        return output

class Parse(Factory):
    def process(self, input):
        print(" ...")
        output = {}
        return output

class Load(Factory):
    def process(self, input):
        print(" ...")
        output = {}
        return output

pipe = {  
    ""   : Extract(),
    "" : Parse(),
    "" : Load(),
}

inputs = {}  
#  
for name, instance in pipe.items():  
    inputs = instance.process(inputs)
      
      



:





 ... 
 ... 
 ...
      
      



for



, . - , . Β« , , , Β» - , - Erlang.





, , , .





(factorial



) (factorial_rec



). . , . .





, , - ;-), .. .





#    
#    

def main():
    #  ( c   )
    pipe(int(input('   : ')),    
         lambda n: (n, reduce(lambda x, y: x * y, range(1, n + 1))),    
         lambda tup: 
             print(f'  {tup[0]}  {tup[1]}'))        

#   
main()
      
      



:





   : 4 (Enter)
  4  24
      
      



- , .





, .. - - . . .





#    
#     

def get_int(msg=''):
    return int(input(msg))

def main():
    #  1.     
    def factorial_rec(n): 
        fn = lambda n, acc=1: acc if n == 0 else fn(n - 1, acc * n)
        return n, fn(n) 
  
    #  2.  
    def factorial(n):     
        return n, reduce(lambda x, y: x * y, range(1, n + 1)) 
   
    #  
    def indata():
        def validate(n):  #   
            if not isinstance(n, int):
                raise TypeError("   .")
            if not n >= 0:
                raise ValueError("   >= 0.")
            return n        
        msg = '   : '
        return pipe(get_int(msg), validate)
   
    #  
    def outdata():
        def fn(data):
            n, fact = data
            print(f'  {n}  {fact}') 
        return fn

    #  ( )
    pipe(indata(),     # : -       : int
         factorial,    # : int     : 
         outdata())    # :   : -    

#   
main()
      
      



:





   : 4 (Enter)
  4  24
      
      



:





pipe(indata(), 
     factorial, 
     outdata())
      
      



, .. indata



, factorial



outdata



. indata



, . factorial



, , , . outdata



. , indata , .





. -, - . -, .





:





  • . , factorial



    , factorial_rec



    .





pipe(indata(), factorial_rec, outdata())
      
      



  • , .





, – . debug



:





def debug(data):
    print(data) 
    return data
      
      



, :





pipe(indata(), debug, factorial, debug, outdata())
      
      



, :





:





   : 4 (Enter)
4
(4, 24)
      
      



4 24





, factorial



4



, (4, 24)



. , , . , debug



-, .





.





#    
#     

def main():
    # 
    def fibonacci(n, x=0, y=1):
        #  fib  n-  .
        fib = lambda n, x=0, y=1: x if n <= 0 else fib(n - 1, y, x + y)
        #  reduce     acc
        acc = []
        reduce(lambda _, y: acc.append(fib(y)), range(n + 1))
        return n, acc

    #   
    def validate(n):         
        if not isinstance(n, int):
            raise TypeError("   .")
        if not n >= 0:
            raise ValueError("    .")
        if n > 10:
            raise ValueError("     10.")
        return n

    #  
    def indata():
        msg = '      10: '
        return pipe(get_int(msg), validate)

    #  
    def outdata():
        def fn(data):
            n, seq = data
            msg = f' {n}   :'
            print(msg) 
            [print(el) for el in seq]
        return fn

    #  ( )
    pipe(indata(), fibonacci, outdata()) 

#   .
main()

 
      10: 10 (Enter)
 10   :
1
1
2
3
5
8
13
21
34
55
      
      



#    
#    
#   

def main():
    # 
    def range_sum(data):  
        seq, params = data
        fn = lambda start, end: 0 if start > end \
                                  else seq[start] + fn(start + 1, end)
        return fn(*params)

    #  
    def indata():
        seq = [1, 2, 3, 4, 5, 6, 7, 8, 9]    
        params = (2,5)   # params -   start, end
        return seq, params  

    #  
    def outdata():
        def f(data):
            msg = '   2  5   '
            print(msg, format(data), sep='') 
        return f

    #  ( )
    pipe(indata(), range_sum, outdata()) 

#   .
main()

 
   2  5   18
      
      



Python . : . . , , , , , . , .)





Github. Strating Out with Python. , . ,





  • PyCon .





  • , .





  • Youtube Β« Python - Β».





, , Python , map/filter/reduce/zip functools. . , , , .








All Articles