Monkey Patch란?

업무를 하면서 가끔 회사에서 Monkey Patch를 하고...라는 말을 듣는다.

무슨뜻이지? 하면서 한번도 찾아본적이 없어서 이번에 찾아보고 정리한다.

 

위키피디아에 이렇게 정리되어 있다.

 

A monkey patch is a way for a program to extend or modify supporting system software locally (affecting only the running instance of the program).

해석을 한다면 monkey patch는 프로그램이 내부 프로그램을 로컬에서 잠시 다른 기능을 수행하도록 수정하거나 확장하도록 하는 방법이며 이는 로컬이므로 런타임시에만 영향을 주는 방식이다.

 

예시로는 Python으로 아래와 같이 있다.

>>> import math
>>> math.pi
3.141592653589793
>>> math.pi = 3
>>> math.pi
3
>>> ================================ RESTART ================================
>>> import math
>>> math.pi
3.141592653589793
>>>

위의 내용만 봐도 이해는 했을 것이다.

그런데 어디다가 쓰지?

 

예상은 했겠지만, Test할때 주로 쓴다.

그리고 다른 기능을 수행하도록 하는 부분 또한 Mock Object를 원하는 기능이 수행하도록 만든 뒤 그것으로 replace하는 것이다.

 

my_calendar.py

import requests
from datetime import datetime

def is_weekday():
    today = datetime.today()
    return (0 <= today.weekday() < 5)

def get_holidays():
    r = requests.get('http://xxx.holiday.org/api/holidays')
    if r.status_code == 200:
        return r.json()
    return None

위와 같은 코드를 만들었고, 이제 테스트를 해야 한다.

get_holidays() 함수는 requests가 필요한데 아직 위의 url에서는 해당 api를 서비스하지 않고 있다.

즉, 테스트를 하기가 어렵다.

 

아래와 같이 tests.py를 한번 보자.

 

import unittest
from my_calendar import get_holidays
from requests.exceptions import Timeout
from unittest.mock import patch

class TestCalendar(unittest.TestCase):
    @patch('my_calendar.requests')
    def test_get_holidays_timeout(self, mock_requests):
            mock_requests.get.side_effect = Timeout
            with self.assertRaises(Timeout):
                get_holidays()
                mock_requests.get.assert_called_once()

if __name__ == '__main__':
    unittest.main()

위의 코드는 my_calenadar class를 Test하기 위한 코드이다.

 

@patch('my_calendar.requests') 이부분이 위에 설명되었던 Monkey Patch부분이다.

@patch decorator에 들어가는 파라메터(my_calendar.requests)는 replace 대상(target)이 되는 속성이다.

즉, my_calendar 소스내에 import requests 가 있는데 이걸 내가 원하는 requests로 바꿀려고 한다.

그럼 바꿀려고 하는 mock object 참조변수는 def test_get_holidays_timeout(self, mock_requests) 의 mock_requests이다.

 

따라서 해당 함수 안에서 mock_requests.get.side_effect = Timeout 즉 해당 request의 get() 함수는 실제 url에 접속하고 관계없이,

Timeout을 Raise해줘.. 라는 것으로 가짜 object를 만들었다. 이제 @patch decorator를 통해 실제 my_calendar.py의 requests 객체를 바꾸었으니 Exception Test가 가능해졌다.

 

with 문 이하는 본 monkey patch와 관계없는 Mock관련 사항이므로 설명을 생략한다.

 

따로 Python의 Mock 관련해서는 설명 글을 게시할 예정이다.

 

정리하자면 위와 같이 requests 는 아시겠지만 이를 테스트할려면 해당 api서버가 개발을 해주거나 서비스를 제공해야 테스트 가능하다.

이부분을 내가 로컬에서 단위테스트를 하기 위해서는 원래의 requests를 내가 원하는 방식으로 바꿔줘야 하는데 이게 monkey patching이며, 원하는 방식으로 바꾸기 위한 임시객체(가짜객체)를 Mock Object라 하며 Python에서는 unittest.mock에서 강력한 기능들을 제공한다.

 

끝.

'Python' 카테고리의 다른 글

Python Celery Task Monitoring  (0) 2019.10.08
Python Lambda  (0) 2019.09.03
Celery from scratch  (0) 2019.05.20
Python Excel to MySQL  (0) 2019.04.12
Database 정보 CSV 작성방법  (0) 2019.04.10

PostgresQL String Concatenation 엔 뭔가 다른 것이 있다.

이 다른 것을 알지 못하면 큰 장애를 유발 할 수 있다.

PostgresQL을 많이 다루어 보지 못해서 생긴 이슈일수도 있다.

꼭 아래 내용을 숙지할 것.

 

String concatenation 2 가지 방법

1. || Operator

||( vertical bar ) 2개를 연속하여 사용하는 방법이며 이것은 NULL 을 인지한다.

따라서 NULL이 들어오면 모든 문자는 NULL이 된다.

select 'A'||NULL; --> NULL Return

Ex) Oracle의 경우, select 'A'||NULL from dual --> 'A' Return 됨.

 

2. concat Function

concat Function은 NULL을 무시한다.

select concat('abc', NULL, 'def'); --> 'abcdef' 즉, NULL을 무시하여 concat('abc','def') 와 같은 결과임.

 

참조 URL) http://www.postgresqltutorial.com/postgresql-concat-function/

 

PostgresQL 에서의 Empty String

소위 말하는 '' 이다.( Single Quatation 2개를 연속해서 씀.)

PostgresQL에서 ''은 NULL 이 아니고 말 그대로 "빈 문자열" 자체의 의미이다.

하지만 다른 DB에서는 NULL 로 인지한다.

따라서 select * from aaa where col1 is null 과 where col1 = '' 은 다르다.

그리고 not null 컬럼에 ''이 들어갈수 있다. 육안으로 보기엔 null처럼 보이니 주의해야 한다.

 

 

대량 엑셀을 효율적으로 ...

배경

현업이 IT부서에 Raw Data를 뽑아달라고 했다.

뽑아보니 레코드 수도 그렇지만 700Mb가 넘는 사이즈였다...

이걸로 뭘 하겠냐고 물어보니 엑셀함수를 이용해서 데이터 검증을 하겠다고 한다...

일단 뽑아줄테니 이걸로 업무하기는 어려울것 같다, 화일 여는데만 10분걸리고 수정/저장할때마다 10분씩 걸릴거다..

라고 하였다. ( 매주 하던 작업이고 할때마다 하루종일 걸린다고 했다..)

 

업무처리 방식 제안

도저히 생산성이 안나올테니 걸어볼 엑셀 함수와 최종 어떤 작업할껀지 알아내고, IT부서에서 기본적인 엑셀 작업을 해주기로 했다.

 

Python 활용..

Python script를 통해 제공한 엑셀을 내 Local DB( MySQL ) 로 로드하여 각종 함수를 적용하고 다시 그것을 엑셀로 뽑아서 주면 된다.

다른 라이브러리도 있지만 나는 pandas 를 익숙하게 잘 쓰고 싶어서 pandas로 하고 python orm은 sqlalchemy를 활용하기로 했다.

 

따로 엑셀함수등 구현은 업무마다 많이 상이하니..

"Excel File을 DB에 로드" 하는 부분만 예시로 기재한다.

import pandas as pd
from sqlalchemy import create_engine
table = pd.read_excel('엑셀화일명.xlsx', sheet_name='쉬트명', header=0,)
engine = create_engine("mysql+pymysql://DB계정:비밀번호@127.0.0.1:3306/스키마명", encoding='utf-8-sig')
table.to_sql(name='테이블명', con=engine, if_exists='append', index=False)

위 Script는 연습해서 외워서 필요할때 테이블만 만들고 간단히 메일쓰듯이 할 수 있으면 좋겠다.

이거 하나 해주니 현업이 엄청 좋아했다. 덩달아 나의 평가도 올라갔다.


잠깐! 꼭 알아야 할 것

1. Encoding

내 Table 의 첫번째 컬럼은 Bigint(20) 이었다.

그런데 들어가다가 오류가 났다. 분명히 눈엔 47748649 만 보이는데...Why?

위 engine 의 encoding option을 보면 처음에는 utf-8 로 했다가 utf-8-sig 로 하니까 잘된다.

utf-8의 BOM문제라고 한다.  그래서 그냥 utf-8 이 아닌 utf-8-sig ( 약어는...signature로 예상됨. )

로 변경하니 잘된다. 따라서 위 오류나면 ( \ufeff --> \u : 유니코드 의미, feff : UTF-16 Big Endian 의미)

즉 UTF-16 Big Endian으로 Encoding된 문자를 utf-8로 해서 앞에 BOM 문자가 포함되어 오류남.

위처럼 하면 utf-8 로 인코딩하되, signature를 보고 input 문자의 encoding을 확인하라는 의미임.

참고 URL : http://blog.wystan.net/2007/08/18/bom-byte-order-mark-problem

 

 2. sqlalchemy , pymysql 등 환경

pip install pandas
pip install pymysql
pip install sqlalchemy

따라하기 예제코드

아래는 Full Code 이며, 계정과 비밀번호, 스키마명만 변경해서 하면 된다.

테이블구조는 예제 실습을 위해 임의로 정하였다.

 

내 Local에 테이블이 없는 상태이다.

mysql table 확인

처리해야 할 엑셀 화일 샘플이다.( raw_data_test.xlsx)

아래 코드 실행

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://계정:비밀번호@127.0.0.1:3306/스키마명", encoding='utf-8-sig')

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.mysql import BIGINT
from sqlalchemy import Column, String, Integer

Base = declarative_base()

class BillRawTest(Base):
    __tablename__ = 'bill_raw_test'
    __table_args__ = {'extend_existing': True}
    bill_id = Column(BIGINT(20), primary_key  = True)
    item = Column(String(20))
    amount = Column(Integer)

metadata = Base.metadata
metadata.create_all(engine)

df = pd.read_excel('~/Downloads/bill_raw_test.xlsx', sheet_name='Sheet1', header=0)
df.to_sql(name='bill_raw_test', con = engine, if_exists='append', index=False)

 

실행 후 DB조회

 

끝.

'Python' 카테고리의 다른 글

Python Celery Task Monitoring  (0) 2019.10.08
Python Lambda  (0) 2019.09.03
Celery from scratch  (0) 2019.05.20
Python - Monkey Patch  (1) 2019.05.02
Database 정보 CSV 작성방법  (0) 2019.04.10

+ Recent posts