20200924 - Python Thread pool, Cache, Mocking snowflake cursor

class SFExecutionPool:
    def __init__(self):
        self.func_li = []

    def add(self, func, **kwargs):
        self.func_li.append((func, kwargs))

    def execute(self):
        with ThreadPool(len(self.func_li)) as pool:
            multiple_results = [
                pool.apply_async(func=func, kwds=kwargs)
                for func, kwargs in self.func_li
            ]
            results = [res.get() for res in multiple_results]
        return results

1. Python Thread pool

유명한 사실이지만 python은 싱글 스레드로 움직이는 인터프리터 언어이며 이에 따라 병렬성을 구현하기 굉장히 어렵다. 따라서 threading module을 활용하여 멀티쓰레딩을 구현해야 한다. 하지만 golang과 같이 일반적인 병렬적 프로그래밍을 구현하기는 어려운데, 그 이유는 GIL이다. GIL에 대한 자세한 내용은 realpython.com/python-gil/

What is the Python Global Interpreter Lock (GIL)? – Real Python

 

그래서 공통된 자원에 대한 접근을 하는 쓰레딩의 경우 GIL이 막는다. 따라서 I/O 작업에 대해 멀티쓰레딩을 활용할때 효과적으로 보인다. 나는 다음과 같이 threading pool을 구현해보았다.

class SFExecutionPool:
    def __init__(self):
        self.func_li = []

    def add(self, func, **kwargs):
        self.func_li.append((func, kwargs))

    def execute(self):
        with ThreadPool(len(self.func_li)) as pool:
            multiple_results = [
                pool.apply_async(func=func, kwds=kwargs)
                for func, kwargs in self.func_li
            ]
            results = [res.get() for res in multiple_results]
        return results

I/O작업들을 하는 function들을 pool 객체의 func_li에 넣어놓고, execute를 하면 thread pool에서 async하게 작업을 실행한다.

 

 

2. Python Cache

반복적인 명령에 대해 cache화를 진행해놓으면 매우 효율적이다. 특히 decorator를 활용해서 구현한다면 아주아주 좋다. 다음과 같이 구현하였다.

cache = LRUCache(**settings.cache.dict())


def lru_cache(func):
    def wrapper(*args, **kwargs):
        key = f"{func.__name__}_{kwargs.values()}"
        results = cache.get(key=key)
        if results is not None:
            return results

        results = func(*args, **kwargs)
        cache.set(key=key, value=results)
        return results

    return wrapper

너.무.좋.아

 

3. Mocking snowflake cursor

요즘 한창 snowflake를 활용하고 있는데, test code를 짤 때 mocking에서 곤란한 지점이 있었다. 왜냐하면 코드가 다음과 같았기 때문이다.

with self._engine.cursor() as c:
	query_result = c.execute(command=query, params=params).fetchall()

engine을 mocking해서 객체를 선언할 때 넣고, 이 동작 자체를 mocking해야 할텐데.. 그래서 mock cursor라는 새로운 class를 만들어 다음과 같이 정의하였다.

 

class MockCursor(object):
    def __init__(self, return_value):
        self.return_value = return_value

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def execute(self, command, params):
        return self

    def fetchall(self):
        return self.return_value

snowflake-python connector 소스코드를 살펴보면 cursor가 execute()함수를 실행하게 될 때 결과값을 return하게 되는 것이 아닌 객체 스스로를 return한다. 그러다가 iterate한 작업에 들어가게 되면 결과값을 뱉어내는 것. 그래서 나도 그와 똑같은 logic으로 mock cursor를 구현하였다. 대신 with ~ as 를 실행하게 될때 __enter__와 __exit__과 같은 context manager가 필요하여 위와 같이 정의하였다. 저렇게 정의하면 내가 원하는 값을 return하게 하는 mock cursor 제작 완성~~

아.이.좋.아~