pyfunctional - seq 오브젝트에 외부 메서드 추가하기

함수형 프로그래밍은 뛰어난 확장성과 안정성을 가지고 있다. 

함수형 패러다임을 가져가면서 실무에서 사용할 수준의 커스텀 메서드를 구현하고 싶다면 도움이 될 것 같다. 

from functional import seq 
from functional.pipeline import extend
from typing import TypeVar
from typing import List, Iterable

T = TypeVar('T')
lst = [1, 2, 3, 4, 5]

 

 

01. 기본 메서드 사용하기

pyfunctional의 seq을 이용하여 함수형 프로그래밍을 구현하면 다음과 같이 내장된 함수를 기본적으로 사용할 수 있다. 

# 01. use built-in map, filter
rs = (
    seq(lst)
    .map(lambda x: x + 1)
    .filter(lambda x: x % 2 == 0)
)

print(rs) # [2, 4, 6]

 

02. extend로 메서드 확장하기

그런데 내장된 함수가 아니면서, sequence 오브젝트에서 메서드 체이닝을 하고싶다면 어떻게 해야할까?

pipeline에서 지원하는 데코레이터인 extend를 쓰면된다. 

extend는 사용자 정의 함수를 seq.method 형태로 체이닝을 할 수 있도록 확장시켜주는 데코레이터이다. 

# 02. use extend functions

@extend
def add(it: Iterable[T], n: int):
    return [x + n for x in it]

@extend
def is_even(it: Iterable[T]):
    return [x for x in it if x % 2 == 0]

rs = (
    seq(lst)
    .add(1)
    .is_even()
)
print(rs)

 

03. 상속받아서 메서드 확장하기 

extend가 아니더라도 seq를 생성해낸 클래스를 직접 상속받아서 메서드를 구현할 수 있다.

from functional.pipeline import Sequence

class MySequence(Sequence):
    def __init__(self, iterable):
        super().__init__(iterable)

    def map(self, func):
        return MySequence(super().map(func))

    def filter(self, func):
        return MySequence(super().filter(func))

    def flat_map(self, func):
        return MySequence(super().flat_map(func))

    def add(self, n: int):
        return self.map(lambda x: x + n)

    def is_even(self):
        return self.filter(lambda x: x % 2 == 0)

my_seq = MySequence(lst)
rs = (
    my_seq
    .add(1)
    .is_even()
)
print(rs)  # [2, 4, 6]

 

 

04. 직접 구현 시 주의사항

함수를 구현할 때 가장 중요한 것은 아래 3가지 인터페이스를 구현하는 것이다.

1. Iterable 타입의 객체를 받는다.

2. Iterable 타입의 객체를 적절히 처리한다.

3. 의도한 타입의 객체를 return한다.

- 데이터를 추가로 처리한다면 iterable을 리턴한다.

- 결과값을 보고자 한다면 scalar를 리턴한다. 

 

부록. pyfunctional extend 함수가 메소드를 추가하는 방법

extend는 정의한 함수를 받을 뿐 아니라, aslist, final 등과 같은 파라미터를 추가로 받아서 seq에 메서드를 등록한다. 

사용자가 final=True로 줄 경우, iterable로 감싼 seq 객체를 리턴하는 것이 아닌, 함수가 처리한 output을 그대로 리턴한다. 

즉, 최종 결과 값이 보장된다면 사용 가능하다.

 

aslist=True로 줄 경우, input으로 들어온 iterable 객체를 list로 변환하여 넣어준다. 

만약 False라면, iterable인 seq 객체를 그대로 보낸다. 

최종적으로 Sequence라는 클래스에 사용자가 정의한 함수명을 key로 하는 함수를 등록한다. 

def extend(func=None, aslist=False, final=False, name=None, parallel=False):    
    if func is None:
        return partial(extend, aslist=aslist, final=final, name=name, parallel=parallel) # func가 None이면 partial을 리턴한다.
    assert func is not None

    @wraps(func)
    def wrapper(self, *args, **kwargs):
        if final:
            return func(self.sequence, *args, **kwargs) # seq객체가 현재 가지고 있는 sequence를 리턴한다.

        if aslist:
            func_ = lambda seq: func(list(seq), *args, **kwargs) # aslist가 True면 iterable을 list로 변환해서 func에 넘긴다.
        else:
            func_ = lambda seq: func(seq, *args, **kwargs) # aslist가 False면 iterable을 그대로 func에 넘긴다.

        transform = transformations.Transformation(
            f"extended[{name or func.__name__}]",
            func_,
            {ExecutionStrategies.PARALLEL} if parallel else None, 
        )
        return self._transform(transform)
    setattr(Sequence, func.__name__, wrapper) # seq객체에 func의 이름을 가진 메소드를 추가한다.
    return wrapper

 

 

wrapper가 리턴하는 self._transform(transform)는 pyfunctional의 내장된 다른 함수를 구현하는 방법과 동일하다.

# 내장 sorted 함수 등록
# Sequence.sorted
def sorted(self, key=None, reverse=False):
	return self._transform(transformations.sorted_t(key=key, reverse=reverse))

# 내장 sorted 함수 로직 구현
# transformations.sorted_t
def sorted_t(key=None, reverse: bool = False):
	return Transformation(
        "sorted", lambda sequence: sorted(sequence, key=key, reverse=reverse), None
    )