python 파이썬 DB(mysql, mssql) 다루기

 

서로 다른 여러가지 서버의 DB를 연동하기에 용이하도록 작업한 내용 정리.

두가지 파일만 만들고 바로 테스트할 수 있다. pymysql, pymssql 설치 필요. pip install pymssql

 

코드

db_config.py

• 접속할 서버들의 connection 정보를 기입해 둔다.
• type 변수에 DB 종류를 기입한다(mysql or mssql).

db_server = {'local':{'type':'mysql', 'host':'127.0.0.1', 'user':'root', 'pass':'abcdefg', 'db':'HOME'}}
db_server['image'] = {'type':'mssql', 'host':'111.222.333.444', 'user':'ddd', 'pass':'123456', 'db':'TA_LINK'}
db_server['test'] = {'type':'mssql', 'host':'111.222.333.444', 'user':'aaa', 'pass':'123456', 'db':'TA_MALL_LINK'}

 

db.py

• DB에 접속하는 함수 db_connect
• DB에 접속 후 cursor 를 생성하는 함수 db_cursor
• INSERT, UPDATE, DELETE 를 처리하는 함수 db_query
• SELECT 를 처리하는 함수 select_all, select_one

import db_config
import pymysql
import pymssql

def db_connect(server):
    db_type = db_config.db_server[server]['type']
    host    = db_config.db_server[server]['host']
    user    = db_config.db_server[server]['user']
    pwd     = db_config.db_server[server]['pass']
    db      = db_config.db_server[server]['db']

    if db_type == 'mysql':
        con = pymysql.connect(host=host, user=user, password=pwd, db=db, charset='utf8')
    elif db_type == 'mssql':
        con = pymssql.connect(server=host, user=user, password=pwd, database=db, charset='utf8')
    
    return con

def db_cursor(server='local'):
    con = db_connect(server)

    db_type = db_config.db_server[server]['type']

    if db_type == 'mysql':
        cur = con.cursor(pymysql.cursors.DictCursor)
    elif db_type == 'mssql':
        cur = con.cursor(as_dict=True)
    
    return con, cur

def db_query(query, server='local'):
    con, cur = db_cursor(server)
    cur.execute(query)

    id = cur.lastrowid

    con.commit()
    con.close()

    return id

def select_all(query, server='local'):
    con, cur = db_cursor(server)
    cur.execute(query)

    rows = cur.fetchall()

    con.close()

    return rows

def select_one(query, server='local'):
    con, cur = db_cursor(server)
    cur.execute(query)
    
    rows = cur.fetchone()
    
    if rows is None:
        rows = {}
    
    con.close()

    return rows

 

특이사항

1. SELECT 결과를 딕셔너리 형태(key - value)로 받아오도록 구현하였다.
   이를 위해 cursor 생성 메소드에 파라미터를 mysq : pymysql.cursors.DictCursor, mssql : as_dict=True 로 지정.
2. db_query 함수는 INSERT 시 생성된 레코드의 id 값(auto_increase)을 얻어온다.
   UPDATE / DELETE 시에는 0 을 반환한다.
3. select_one 함수는 결과값이 없는 경우의 처리를 하고 있는데 상황에 따라 변형이 필요해 보인다.

 

테스트

test.py

import db

# 여러 레코드를 SELECT하여 반복문으로 돌리기, 각각의 칼럼에 접근.
# 두번째 인자를 생략하였으므로 local 서버에 질의.
row = db.select_all("SELECT * FROM sj_code_rival LIMIT 2")

i = 0

while i < len(row):
    print(row[i].items())
    print(row[i]['ORDER_NO'])
    i += 1

# 결과
# dict_items([('ORDER_NO', 'BCFF01301'), ('PRICE', 160000), ('TRAN_DATE', datetime.datetime(2019, 2, 28, 11, 2, 35, 780000)), ('RCV_DATE', None)])
# BCFF01301
# dict_items([('ORDER_NO', 'AAGD22193'), ('PRICE', 486000), ('TRAN_DATE', datetime.datetime(2018, 7, 10, 11, 2, 35, 780000)), ('RCV_DATE', None)])
# AAGD22193

# 하나의 row 만 불러오는 경우.
# 두번째 인자를 지정하여 test 서버에 질의.
row = db.select_one("SELECT * FROM sj_code_rival WHERE ORDER_NO='BCFF01301'", "test")

print(row['PRICE'])

# 결과
# 160000

# INSERT / UPDATE / DELETE 쿼리
print(db.db_query("UPDATE lt_counter SET hit='100' WHERE ip='1.1.1.1'"))

# 결과
# 0

 

문자열, 데이터 타입 처리

NULL & Nonetype 처리

딕셔너리 타입으로 처리하다 보니 실제로 돌려보면 한가지 처리가 필요하다.

칼럼 값이 공백 '' 이 아니라 실제로 NULL 인 경우에는 None 이라 출력되고, 칼럼 자체가 존재하지 않으면 에러가 발생한다.

따라서 get 함수를 써야 한다.

print(row['SANGHO'])
# 대신
print(row.get('SANGHO', '')) # NULL이거나 해당 칼럼이 없으면 공백으로 치환

 

datetime 처리

datetime 또는 date 타입 등 날짜 칼럼의 값을 가져오는 경우 처리해야 할 부분이 있다.

 

데이터타입	MSSQL	변환하고자 하는 형식
datetime	2023-02-28 11:02:35.780000	2023-02-28

%Y-%m-%d %H:%M:%S 와 같은 형식으로 변환하고자 할 경우 NULL 인 경우까지 대비해야 한다.

row[i]['RCV_DATE'] = row[i].get('RCV_DATE', '')[0:10] if row[i]['RCV_DATE'] else ''
또는
row[i]['RCV_DATE'] = row[i]['RCV_DATE'].strftime('%Y-%m-%d') if row[i]['RCV_DATE'] else ''