[ 목적 ]
회사에서 데이터 마이그레이션 업무 중 파이썬으로 스크립트를 만들어야하는 경우가 있었다. 대학교 연구실에서 크롤링, 데이터 분석, 신경망을 공부하던 시절에 파이썬을 많이 사용해봤지만, 자바를 공부하기 전 사용을 했어서 불편함을 못느꼈다.
다른 프로젝트에서 타입 스크립트를 사용하고 회사에서는 자바를 주 언어로 사용하다 보니 파이썬을 다시 사용하려고 하니까 타입 힌팅, 코드 가독성이 너무 떨어져서 효율이 안 나오고 있었다.
지금도 수준이 낮지만 그때는 진짜 애기 개발자여서 그냥 그런가보다 싶었는데, 이제는 좀 효율적이게 코딩을 해보고 싶어서 방법을 찾던 중 typing 모듈을 찾게 되었다.
[ typing 모듈 ]
파이썬은 인터프리터 언어이며 동적 타입 언어이다. 그래서 IDE는 변수에 어떤 자료형이 할당될지 알지 못하므로 IDE를 사용하더라도 타입 힌팅이 제한적으로 지원한다. 이러한 문제점을 해결하기 위해 자료형을 선언을 하지만 복잡한 구조의 객체일 경우에는 이 마저도 힘들다. 그러므로 typing과 같은 라이브러리를 사용하여 자료형 alias를 설정해주어야한다.
아래 링크가 파이썬 공식 Document이다.
https://docs.python.org/ko/3/library/typing.html
typing — Support for type hints
Source code: Lib/typing.py This module provides runtime support for type hints. For the original specification of the typing system, see PEP 484. For a simplified introduction to type hints, see PE...
docs.python.org
typing 모듈이란 파이썬 3.5부터 표준 라이브러리에 추가되었다.
개발자가 새로운 타입을 선언하여 사용할 수 있고, 복잡한 구조의 객체일 경우 타입 선언을 통해 타입 힌팅을 지원해 줄 수 있다.
[ 사용법 ]
사용법은 간단하다. 라이브러리를 import하고 자신이 원하는 타입을 선언하면 된다. 예시를 들어보자.
class TestObj:
def __init__(self):
self.a = "a"
self.b = "b"
self.typeHint = "hint"
a: List[TestObj] = []
b = []
for i in range(10):
tmp = TestObj()
a.append(tmp)
for i in range(10):
tmp = TestObj()
b.append(tmp)
for i in a:
print(i)
for i in b:
print(i)위와 같은 코드가 있을 때, a와 b를 반복문을 돌리고 있다. 그렇다면 차이점은 무엇일까?
둘 다 똑같이 TestObj를 배열에 넣어주고 있다. 하지만 a변수에는 List타입에 TestObj가 있다는 것을 개발자가 명시적으로 알려주고 있다.
맨 아래 a와 b를 반복문 돌릴때 'i. '을 찍어보면 알 수 있다. a를 반복문 돌릴 경우에는 타입 힌팅이 노출되지만 b일 경우에는 노출되지 않는다.
이는 곧 개발 생산성과 코드 가독성으로 이어지므로 중요하다.위와 같은 간단한 코드일 경우에는 괜찮겠지만 조금만 더 복잡해지면 개발 생산성이 대폭 낮아진다.
[ 중요성 ]
아래 코드는 typing을 적용하기 전 코드이다. 딱 봐도 클래스의 속성에 뭐가 뭔지 잘 모를 것이다.
요약해서 설명해보면, Parser 클래스가 DB를 조회하면서 테이블의 레코드들을 TmpDB에 넣어두고 pickleing하는 코드이다.
Parser는 source와 target 2개의 parser돌아가고 두 parser를 비교해서 target DB에 넣어주는 코드를 개발하고 있었던 중이었다.
paresr에서 tmpdb를 순회할때 코드가 다음과 같이 되었었다.
target.tmpDB.cmpTmpUpdate[source.tmpDB.cmpTemplate.tmp_upt_id]
위와 같이 속성들을 따라가야할 경우에 typing 라이브러리가 유용하다.
개발자도 사람인데 저 모든 속성을 어떻게 외우고 있겠는가?
class CmpTemplate:
def __init__(self, tmp_id, title, desc, system_flg, del_tm, lock_flg, tmp_grp_id, tmp_upt_id, priority_type, order_no, hash_id, job_flg, insp_flg, ci_flg, mtr_flg):
self.tmp_id = tmp_id
self.title = title
self.desc = desc
self.system_flg = system_flg
self.del_tm = del_tm
self.lock_flg = lock_flg
self.tmp_grp_id = tmp_grp_id
self.tmp_upt_id=tmp_upt_id
self.priority_type=priority_type
self.order_no=order_no
self.hash_id=hash_id
self.job_flg=job_flg
self.insp_flg=insp_flg
self.ci_flg=ci_flg
self.mtr_flg=mtr_flg
def __str__(self):
return f"CmpTemplate(tmp_id={self.tmp_id}, title={self.title}, desc={self.desc}, system_flg={self.system_flg}, del_tm={self.del_tm}, lock_flg={self.lock_flg}, tmp_grp_id={self.tmp_grp_id}, tmp_upt_id={self.tmp_upt_id}, priority_type={self.priority_type}, order_no={self.order_no}, hash_id={self.hash_id}, job_flg={self.job_flg}, insp_flg={self.insp_flg}, ci_flg={self.ci_flg}, mtr_flg={self.mtr_flg})"
class CmpUpdate:
def __init__(self, tmp_id, con_type_flg, script_method, script_type, api_ctg, tmp_upt_id, priority_type, os_type, user_id, img_id, upt_tm):
self.tmp_id = tmp_id
self.con_type_flg = con_type_flg
self.script_method = script_method
self.script_type = script_type
self.api_ctg = api_ctg
self.tmp_upt_id=tmp_upt_id
self.priority_type=priority_type
self.os_type=os_type
self.upt_tm =upt_tm
self.img_id=img_id
self.user_id=user_id
def __str__(self):
return f"TmpUpdate(tmp_id={self.tmp_id}, con_type_flg={self.con_type_flg}, script_method={self.script_method}, script_type={self.script_type}, api_ctg={self.api_ctg}, tmp_upt_id={self.tmp_upt_id}, priority_type={self.priority_type}, os_type={self.os_type}, user_id={self.user_id}, img_id={self.img_id}, upt_tm={self.upt_tm})"
class CmpImage:
def __init__(self, img_id,title,img_path,img_data):
self.img_id=img_id
self.title=title
self.img_path=img_path
if img_data:
self.img_data = img_data.tobytes()
def __str__(self):
return f"CmpImage(img_id={self.img_id}, title={self.title}, img_path={self.img_path}, img_data={self.img_data})"
class CmpProp:
def __init__(self,tmp_upt_id, prop_order, prop_title, prop_opt_key, prop_type, prop_required, prop_default_val):
self.tmp_upt_id=tmp_upt_id
self.prop_order=prop_order
self.prop_title=prop_title
self.prop_opt_key=prop_opt_key
self.prop_type=prop_type
self.prop_required=prop_required
self.prop_default_val = prop_default_val
def __str__(self):
return f"CmpProp(tmp_upt_id={self.tmp_upt_id}, prop_order={self.prop_order}, prop_title={self.prop_title}, prop_opt_key={self.prop_opt_key}, prop_type={self.prop_type}, prop_required={self.prop_required}, prop_default_val={self.prop_default_val})"
class CmpGroup:
def __init__(self,tmp_grp_id, grp_name, del_tm, tmp_type, order_no, p_tmp_grp_id):
self.tmp_grp_id=tmp_grp_id
self.grp_name=grp_name
self.del_tm=del_tm
self.tmp_type=tmp_type
self.p_tmp_grp_id=p_tmp_grp_id
self.order_no=order_no
def __str__(self):
return f"TmpGroup(tmp_grp_id={self.tmp_grp_id}, grp_name={self.grp_name}, del_tm={self.del_tm}, tmp_type={self.tmp_type}, order_no={self.order_no}, p_tmp_grp_id={self.p_tmp_grp_id})"
class TmpDB:
def __init__(self):
self.cmpTemplate = {}
self.cmpTmpUpdate = {}
self.cmpImages = {}
self.cmpProps ={}
self.cmpGroups = {}
self.tmpIdList = []
self.tmpUptIdList = []
self.tmpImageIdList = []
self.tmpGrpIdList = []
self.tmpHashIdList = []
class ImportUtil:
def __init__(self, source, target):
"""_summary_
Args:
source (_type_): 기술부 템플릿
target (_type_): 업그레이드 서버 템플릿
"""
self.source = source
self.target = target
self.updateDB = {}
self.insertDB = {}
self.hardCode = ["TS011","TS012"]
def insertData(self):
"""
하나씩 넣어야되고, 시퀸시를 바로바로 가져와서 넣어야댐. 가져오기 참조
"""
query = ""
def updateData(self):
"""
새로 넣을 필요는 없고 업데이트가 필요한 템플릿만 업데이트 시킴
"""
query = ""
def checkSourceData(self):
"""
1. 소스에서 타겟에서 없는 데이터 검색
2. 없는 소스는 그대로 넣어줌
3. 있는 소스들은 변경이력 검색해야됨
4. 조건에 맞는 템플릿 아이디만 모아둠
비교 방법
1. hash_id로 1차 비교.
2. upt_tm으로 비교.
문제점
1. 시스템 템플릿 & 새로운 유형의 템플릿은 사용자가 수정할 수 없으니 UPT_TM이 기술부 쪽이 최신이면 덮어씀
자료구조
- 시간복잡도 1인 hashmap을 주로 이용함
- 업데이트가 필요한 데이터와 신규로 넣어야하는 데이터 2가지 hashmap으로 나눔
"""
# 넣어야하는 템플릿 추출
for i in self.source.tmpDB.cmpTemplate.keys():
if(i not in self.target.tmpDB.cmpTemplate):
self.insertDB[i] = self.target.tmpDB.cmpTemplate[i]
# 업데이트해야하는 템플릿 추출
for i in self.source.tmpDB.cmpTemplate.keys():
if(i in self.target.tmpDB.cmpTemplate):
tmp = self.source.tmpDB.cmpTemplate[i]
target = self.target.tmpDB.cmpTemplate[i]
# 기술부 템플릿
uptId = self.source.tmpDB.cmpTemplate[i].tmp_upt_id
sourceCmpUpdate = self.source.tmpDB.cmpTmpUpdate[uptId]
# 타켓 템플릿
uptId = self.target.tmpDB.cmpTemplate[i].tmp_upt_id
targetCmpUpdate = self.target.tmpDB.cmpTmpUpdate[uptId]
if(sourceCmpUpdate.upt_tm > targetCmpUpdate.upt_tm):
self.updateDB[i] = self.source.tmpDB.cmpTemplate[i]
def reCategorizeTarget(self):
"""
재분류하는 작업
1. 하나씩 포문 돌림
2. 하드코딩된 배열에 값이 없고, 플래그가 0이면 플래그를 2로 설정.
3. tmp_id에 S가 들어가는 컴포넌트 템플릿만 해당함.
4. 이미 재분류되어 타입이 2인것은 제외.
5. 재분류 되었으면 DB값을 업데이트 시켜야함.
"""
Skey = self.source.keys()
Tkey = self.source.keys()
for k in Skey:
if ("S" in cur.tmp_id):
cur = self.source[k]
cur.system_flg = "2"
for k in Tkey:
if ("S" in cur.tmp_id):
cur = self.target[k]
cur.system_flg = "2"
class Parser:
def __init__(self, ip):
self.CONFIG = {
'dbname': 'mccs',
'user': 'mccs',
'password': 'password',
'host': ip,
'port': '5433',
}
self.conn = psycopg2.connect(**self.CONFIG)
self.cursor = self.conn.cursor(cursor_factory=DictCursor)
self.tmpDB = TmpDB()
self.tmpIdList = set()
self.tmpUptIdList = set()
self.tmpImageIdList = set()
self.tmpGrpIdList = set()
self.tmpHashIdList = set()
def getWfCmpTemplate(self):
query = """
select tmp_id, title, "desc", system_flg, del_tm, lock_flg, tmp_grp_id, tmp_upt_id,
priority_type, order_no, hash_id, job_flg, insp_flg, ci_flg, mtr_flg
from public.cmp_templates c
where
c.del_tm is null and
c.tmp_id like 'T%' and
c.tmp_id != 'TS011' and
c.tmp_id != 'TS012'
"""
self.cursor.execute(query)
res = self.cursor.fetchall()
for d in res:
# print(d)
obj = CmpTemplate(**d)
self.tmpDB.cmpTemplate[obj.hash_id] = obj
self.tmpIdList.add(obj.tmp_id)
self.tmpUptIdList.add(obj.tmp_upt_id)
self.tmpHashIdList.add(obj.hash_id)
self.tmpGrpIdList.add(obj.tmp_grp_id)
# print(obj.tmp_grp_id)
def getWfCmpTmpUpdates(self):
placeholders = ', '.join(['%s'] * len(self.tmpUptIdList))
query = f"""select tmp_id, con_type_flg, script_method, script_type, api_ctg, tmp_upt_id, priority_type, os_type, user_id, img_id, upt_tm from public.cmp_template_updates ctu where ctu.tmp_upt_id IN ({placeholders})"""
self.cursor.execute(query, tuple(self.tmpUptIdList))
res = self.cursor.fetchall()
for d in res:
# print(d)
obj = CmpUpdate(**d)
self.tmpDB.cmpTmpUpdate[obj.tmp_upt_id] = obj
self.tmpImageIdList.add(obj.img_id)
def getWfCmpImages(self):
placeholders = ', '.join(['%s'] * len(self.tmpImageIdList))
query = f"""select img_id, title, img_path, img_data from public.cmp_images ci WHERE ci.img_id IN ({placeholders})"""
self.cursor.execute(query, tuple(self.tmpImageIdList))
res = self.cursor.fetchall()
for d in res:
# print(d)
obj = CmpImage(**d)
self.tmpDB.cmpImages[obj.img_id] = obj
#print(obj.img_data)
def getWfCmpGroup(self):
placeholders = ', '.join(['%s'] * len(self.tmpGrpIdList))
query = f"""select tmp_grp_id, grp_name, del_tm, tmp_type, order_no, p_tmp_grp_id from public.cmp_template_groups ctg where ctg.tmp_grp_id IN ({placeholders})"""
self.cursor.execute(query, tuple(self.tmpGrpIdList))
res = self.cursor.fetchall()
for d in res:
obj = CmpGroup(**d)
self.tmpDB.cmpGroups[obj.tmp_grp_id] = obj
def getWfCmpProp(self):
placeholders = ', '.join(['%s'] * len(self.tmpUptIdList))
query = f"""select tmp_upt_id, prop_order, prop_title, prop_opt_key, prop_type, prop_required, prop_default_val from public.cmp_template_properties ctp where ctp.tmp_upt_id IN ({placeholders})"""
self.cursor.execute(query, tuple(self.tmpUptIdList))
res = self.cursor.fetchall()
for d in res:
# print(d)
obj = CmpProp(**d)
if obj.tmp_upt_id not in self.tmpDB.cmpProps :
self.tmpDB.cmpProps[obj.tmp_upt_id] = []
self.tmpDB.cmpProps[obj.tmp_upt_id].append(obj)
else:
self.tmpDB.cmpProps[obj.tmp_upt_id].append(obj)
def close(self):
if self.cursor:
self.cursor.close()
self.cursor = None
if self.conn:
self.conn.close()
self.conn = None
def parsing(self):
self.getWfCmpTemplate()
self.getWfCmpTmpUpdates()
self.getWfCmpImages()
self.getWfCmpGroup()
self.getWfCmpProp()
self.close()