[dummyDataMkaer] SQLAlcehmy를 이용한 더미데이터 생성 프로그램 만들기 with MySQL

해당 프로젝트는 OZ코딩스쿨 수준별반 수업(심화반)을 통해 제작되는 프로젝트의 해결하는 과정의 기록입니다.

진행 환경

OS : Intel Mac Ventura
IDE : Visual Studio Code 

주제 : 테스트를 위한 dummy data 만들기(MySQL)

상황

• MySQL의 Database에 14개의 테이블을 생성함.
• Table만 만든 상태라서 테스트를 위해 더미데이터를 테이블 별로 1000개~2만개 사이로 세팅해달라는 요청을 받음
• 테이블 별로 데이터를 몇 건 씩 세팅할지 지정할 수 있어야함.
• 더미데이터 생성시마다 데이터를 전부 삭제하고 새로 세팅할지, 추가할지를 지정할 수 있어야함.

가이드 (힌트)

• sqlalchemy의 MetaData()를 이용하세요. metadata.reflect() 을 이용하면 현재 접속한 database의 table, column 정보를 얻을 수 있습니다.
• 반복되는 똑같은 행위를 2번 해야한다.
• 1차때처럼 테이블 단위로 칼럼별로 하드코딩하는 형태로 대응하면 계속 수정해야한다.
• 테스트 데이터를 생성하는 코드를 완전 공통화 해보는건 어떨까? ex) AUTO_INCREMENT 칼럼은 대상칼럼에서 제거

사용해야할 라이브러리

sqlalchemy

SQLAlchemy

Faker

Welcome to Faker’s documentation! — Faker 25.1.0 documentation

해보면 좋은 것

파이썬 실행환경은 너무 다양하다.
global 환경에서 실행할수도있고, virtualenv를 직접세팅해서 쓸수도, conda를 쓸수도있고, colab에서 개발하수도있고 천차만별.
가장 최신의 poetry를 기반으로 환경을 세팅해보자!

유의사항

• faker, sqlalchemy 같이 처음 접하는 라이브러리라도 사용법을 찾아서 적용해보는 연습을 하는 시간이라고 생각해주시면됩니다. 사용법을 찾아서 적용하기 어려워도 괜찮습니다. 어려운게 당연합니다.
• 아마존은 같은일을 3번 반복하면 자동화한다고 합니다.
• 파이썬만큼 빠르게 작성하고 실행할수있는 좋은 도구가 많지않습니다.
• 당장 눈앞의 과제를 해결하는 것도 좋은데, 같은일이 반복될거같으면 공통화 해보세요.

이번 과제의 목적

• 테스트 데이터를 매번 세팅하는 일은 아주 번거롭습니다. 더미데이터를 세팅하는 프로그램은 한번 잘 만들어 두면, 실무에서 두고두고 써먹을 수 있는 아주 좋은 도구입니다.
• 이 로직을 잘 정리해서 추후에 배울 서버 API 요청에서 호출하도록하면, 지정한 데이터베이스의 테이블에 더미데이터를 생성해주는 백엔드 API를 만들게 되는 것입니다. 사내 어드민 도구에서 버튼 클릭으로 제어하게 할수 있는겁니다.

진행 내용

데이터베이스 생성

데이터 베이스 생성(데이터 베이스 구조 및 DDL)은 강사님이 제공해주셨습니다.
기반이 되는 데이터 베이스는 아래와 같습니다.(airport DB)

MySQL :: Setting Up the airportdb Database :: 2 Introduction

DDL코드는 아래 깃에 있습니다.

OZ_Backend_School_SideProject/dummyMaker_MYSQL_Refactoring/setting/mysqlDDL/airportdb_DDL.sql at dev · JoonHoSeong/OZ_Backend_S

개발환경 구성하기

poetry 설치

brew install poetry

필요한 라이브러리 설치

poetry add sqlalchemy
poetry add faker
poetry add mypysql
poetry add ipykernel #테스트를 위한 jupyternotebook을 위해 설치

개발 순서

데이터베이스와 연결

import json
import os
from sqlalchemy import create_engine #데이터 베이스와 연결하기 위한 라이브러리
from sqlalchemy import MetaData #메타데이터를 받아오기 위한 라이브러리
from sqlalchemy.engine import reflection #데이터 베이스 내의 정보를 가져오기 위한 라이브러리

with open('setting/config.json') as json_file: #연결 정보 가져오기
    connectData = json.load(json_file)
del json_file
connection_string = f'{connectData['mysql-login']['db_type']}://{connectData['mysql-login']['user_name']}:{connectData['mysql-login']['user_pw']}@{connectData['mysql-login']['db_server']}:{connectData['mysql-login']['db_port']}/{connectData['mysql-login']['db_name']}?charset={connectData['mysql-login']['charset']}'
#데이터 베이스와 연결하기
engine = create_engine(connection_string, echo=False)
#메타데이터 가져오기
metadata = MetaData()
# 데이터베이스에서 테이블 정보 반영
metadata.reflect(bind=engine)
# 터미널 내용 삭제
os.system('clear')

setting의 config.json에 있는 연결 정보를 가져와서 생성된 데이터 베이스와 연결을 해준다.
연결시 메타데이터를 가져와서 추후 데이터 베이스의 내용을 sqlalchemy에서 사용할 수 있도록 해준다.

작동을 위한 CRUD 함수 작성하기

전체 삭제는 쓰지 않기 때문에 사용 안함

import sqlalchemy as db
from sqlalchemy import text
import os

def insert_dummy_data(selected_table, metadata, engine, data) :
    table = db.Table(selected_table, metadata, autoload=True, autoload_with=engine)
    insert_num = 0
    for d in data :
        try :
            query = table.insert().values(d)
            conn = engine.connect()
            conn.execute(query)
            insert_num += 1
            conn.commit()
        except :
            continue
    # os.system('clear') # 터미널 내용 삭제
    if insert_num == 0 :
        return print("제약조건으로 인해 데이터를 더 추가 할 수 없습니다.")
    return print(f"{insert_num}개의 데이터를 {selected_table} 넣었습니다.")

def truncate_table(selected_table, engine) :
    conn = engine.connect()
    conn.execute(text(f"delete from {selected_table};"))
    conn.execute(text(f"alter table {selected_table} AUTO_INCREMENT = 1;"))
    conn.commit()
    # os.system('clear') # 터미널 내용 삭제
    print(f"{selected_table} 내 데이터들을 삭제했습니다.")
        
        
def select_all_data(selected_table, engine) :
    with engine.begin() as conn :
        # print(text(f"select * from {selected_table};"))
        result = conn.execute(text(f"select * from {selected_table};"))
    # os.system('clear') # 터미널 내용 삭제
    print(f"{selected_table}의 데이터")
    for i in result :
        print(i)

테이블과 각 컬럼 정보 가져오기

#각 테이블별 제약 조건을 가져오는 함수
def get_table_detail(tables, inspector): 
    table_detail = dict()
    #정보 및 제약조건 가져오기
    for table in tables :
        if table not in table_detail.keys() : 
            table_detail.update({table : {}})
        columns = inspector.get_columns(table)
        temp = []
        for column in columns:
            # Unique 제약 확인
            unique_constraints = inspector.get_unique_constraints(table)
            if any(column['name'] in uc['column_names'] for uc in unique_constraints) :
                column['unique'] = True
            else :
                column['unique'] = False
            temp.append(column)
        table_detail.update({table : {'details' : temp}})
        del temp
    return table_detail

가져온 제약조건을 이용해서 더미 데이터 생성하기

# MySQL 자료형별 더미 데이터 생성 함수
def generate_dummy_data(data_dict, fake):
    if data_dict['unique'] :
        fake = fake.unique
    try :        
        if data_dict['data_type'] == 'VARCHAR':
            return fake.pystr(max_chars=data_dict['data_length'])
        elif data_dict['data_type'] =='CHAR':
            return fake.pystr(max_chars=data_dict['data_length'])
        elif data_dict['data_type'] =='TEXT':
            return fake.paragraph()
        elif data_dict['data_type'] =='INTEGER'or data_dict['data_type'] =='MEDIUMINT' or data_dict['data_type'] =='SMALLINT':
            return fake.pyint()
        elif data_dict['data_type'] =='FLOAT' or data_dict['data_type'] =='DOUBLE':
            return fake.pyfloat(left_digits=2, right_digits=2, positive=True)
        elif data_dict['data_type'] =='DECIMAL':
            return fake.pydecimal(left_digits=data_dict['data_left_digits'], right_digits=data_dict['data_right_digits'], positive=True)
        elif data_dict['data_type'] =='DATE':
            return fake.date()
        elif data_dict['data_type'] =='TIME':
            return fake.time()
        elif data_dict['data_type'] =='DATETIME' or data_dict['data_type'] =='TIMESTAMP':
            return fake.date_time()
        elif data_dict['data_type'] =='YEAR':
            return fake.year()
        elif data_dict['data_type'] =='BOOLEAN':
            return fake.boolean()
        elif data_dict['data_type'] =='ENUM':
            return fake.random_element(elements=data_dict['data_options'])
    # elif data_dict['data_type'] =='SET':
    #     return fake.random_elements(elements=('Value 1', 'Value 2', 'Value 3'), unique=True)
        elif data_dict['data_type'] =='BLOB' or data_dict['data_type'] =='BINARY':
            return fake.binary(length=10)
        else:
            return None
    except :
        return 'unique_constraints'
    
# 더미데이터 생성 함수
# 각 컬럼의 속성 값들을 이용해서 더미데이터를 생성할수 있는 정보 딕셔너리를 만들고 이를 기반으로 더미데이터 생성
def create_dummy_data_list(table_name, dummy_num, table_detail) :
    datas = []
    fake = Faker()
    for _ in range(dummy_num) :
        temp = {}
        for row in table_detail[table_name]['details'] :
            data_dict = row.copy()
            try :
                if data_dict['autoincrement'] :
                    continue
            except :
                pass
            data = row['type']
            data_dict['data_type'] = data.__class__.__name__
            if data_dict['data_type'] == 'CHAR' or data_dict['data_type'] == 'VARCHAR' :
                # data_dict['data_charset'] = data.charset
                # data_dict['data_collation'] = data.collation
                data_dict['data_length'] = data.length
            elif data_dict['data_type'] == 'DECIMAL' :
                data_dict['data_left_digits'] = data.precision - data.scale
                data_dict['data_right_digits'] = data.scale
            elif data_dict['data_type'] == 'ENUM' :
                data_dict['data_options'] = data.enums
            # elif data_type == 'INTEGER' or data_type == 'MEDIUMINT':
            #     data_dict['data_unsigned'] = data.unsigned
            elif data_dict['data_type'] == 'TINYINT':
                data_dict['data_display_width'] = data.display_width
            # elif data_type == 'TEXT' :
            #     data_dict['data_charset'] = data.charset
            #     data_dict['data_collation'] = data.collation
            del data_dict['type']
            dummy_data =generate_dummy_data(data_dict, fake)
            if dummy_data == 'unique_constraints' :
                return datas
            temp[data_dict['name']] =  dummy_data
        datas.append(temp)
    del temp
    return datas

생성을 위한 메인 함수 만들기

import json
import os
from sqlalchemy import create_engine #데이터 베이스와 연결하기 위한 라이브러리
from sqlalchemy import MetaData #메타데이터를 받아오기 위한 라이브러리
from sqlalchemy.engine import reflection #데이터 베이스 내의 정보를 가져오기 위한 라이브러리
from faker import Faker #더미데이터만들기 위해 사용

from tools.control_table import * #데이터를 DB에 적용
from tools.create_dummy_data import * #더미데이터 생성


def main() :
    with open('setting/config.json') as json_file: #연결 정보 가져오기
        connectData = json.load(json_file)
    del json_file

    with open('command.json') as json_file: #연결 정보 가져오기
        commandData = json.load(json_file)
    del json_file

    connection_string = f'{connectData['mysql-login']['db_type']}://{connectData['mysql-login']['user_name']}:{connectData['mysql-login']['user_pw']}@{connectData['mysql-login']['db_server']}:{connectData['mysql-login']['db_port']}/{connectData['mysql-login']['db_name']}?charset={connectData['mysql-login']['charset']}'
    #데이터 베이스와 연결하기
    engine = create_engine(connection_string, echo=False)
    #메타데이터 가져오기
    metadata = MetaData()
    # 데이터베이스에서 테이블 정보 반영
    metadata.reflect(bind=engine)
    # 터미널 내용 삭제
    # os.system('clear')
    # 데이터베이스 인스펙터 객체 생성
    inspector = reflection.Inspector.from_engine(engine)

    #테이블 이름 가져오기
    tables = inspector.get_table_names()

    table_detail = get_table_detail(tables, inspector)
    for table_name in commandData['table_names'].split(', ') :
    # for table_name in tables : #For Test
        if table_name not in tables :
            print(f'{table_name}테이블은 데이터 베이스에 존재하지 않는 테이블 이름 입니다.')
            continue
        
        if commandData['command'] == 'insert' :
            print(f'{table_name}테이블 더미데이터 생성을 시작합니다.')
            dummy_data = create_dummy_data_list(table_name,commandData['dummy_num'], table_detail)
            print(f'더미데이터 {len(dummy_data)}건 생성 완료')
            insert_dummy_data(selected_table=table_name, engine=engine, metadata=metadata, data=dummy_data)
        elif commandData['command'] == 'new_insert' :
            truncate_table(selected_table=table_name, engine=engine)
            print(f'{table_name}테이블 초기화 완료.')
            print(f'{table_name}테이블 더미데이터 생성을 시작합니다.')
            dummy_data = create_dummy_data_list(table_name,commandData['dummy_num'], table_detail)
            insert_dummy_data(selected_table=table_name, engine=engine, metadata=metadata, data=dummy_data)
            print(f'더미데이터 {len(dummy_data)}건 생성 완료')
        elif commandData['command'] == 'view' :
            select_all_data(selected_table=table_name, engine=engine)
        else :
            print('정해진 실행 방식외 작동을 시도했습니다.\n프로그램을 종료합니다.')
            
if __name__ == '__main__' :
    main()