대신증권 CYBOS PLUS 프로그램 구현 (15) - 차트 데이터 저장하기 ②

프로그램 구현 목표

• 파이참에서 MySQL에 데이터베이스 생성하기
• 파이썬의 Magic Methods 활용하기
• 차트 데이터를 저장할 클래스 생성하기

---

지난 게시글에서는 ①데이터베이스의 존재 여부를 확인하는 기능과 ②데이터베이스와 연결하는 기능을 구현하였다. 이번 게시글에서부터는 데이터베이스를 생성하는 기능을 구현하고, 프로그램 시작과 동시에 해당 데이터베이스의 존재 여부를 판단하고 존재하지 않으면 데이터베이스를 생성하도록 하는 기능을 구현할 예정이다.

 

파이참에서 MySQL에 데이터베이스 생성하기

데이터베이스를 생성할 함수를 con_mysql.py 파일의 `class check_db` 클래스 내부에 `def _make_db(self, db_name):`이라는 이름으로 하여 생성해주도록 하자. 이제 이 함수의 인자로 전달받은 `db_name`이라는 이름의 데이터베이스를 생성하도록 하는 기능을 구현할 것인데, 데이터베이스를 생성하기 위한 명령어는 무엇이 있을까 ? 바로 `execute`이다. 이 명령어는 단순하게 그 뜻을 해석해보면 "실행하다"라는 의미가 있는데, 그 뜻에서도 유추가 가능하겠지만 어떠한 내용을 전달해주고 그 내용을 실행하도록 하는 명령어이다. 여기서 전달해줄 수 있는 내용과 관련해서는 다양한 기능을 수행할 수 있는 만큼 수많은 내용들이 있겠지만, 기본적으로 데이터베이스를 생성하기 위해 전달해주어야 하는 내용은 "CREATE DATABASE 데이터베이스 이름"이다.

이제 이를 _make_db 함수 하단에 작성해볼 예정인데, 이 명령어는 기본적으로 MySQL과 연결하도록 하는 `connect()` 메서드가 호출된 이후에 사용할 수 있다. 다시 말해 `create_engine` 함수를 통해 우리가 연결하고자 하는 데이터베이스(`db_name`)와의 통신이 가능하도록 하는 인스턴스를 생성해두었었는데, 이 인스턴스를 대상으로 `connect()` 함수를 사용하여 파이썬과 MySQL을 연결시킨 이후에 `execute()` 메서드를 통해 데이터베이스를 생성하라는 내용을 전달하고 실행시키도록 해야 한다는 것이다. 이 부분이 이해가 갔다면 아래와 같이 코드를 구현한 후, 프로그램을 실행시켜보면 MySQL WorkBench 내부에 test라는 데이터베이스가 생성된 것을 확인할 수 있다.
※ Line: 26 ~ 28

## con_mysql.py ##

## DataBase
import sqlalchemy as sqlal

class check_db:
    def __init__(self):
        self.engine = sqlal.create_engine(f'mysql+mysqldb://root:비밀번호@127.0.0.1:3306', encoding='utf-8')
        insp = sqlal.inspect(self.engine)
        self.DB_list = insp.get_schema_names()
        print(self.DB_list)

    def con_db(self, db_name):
        if self._if_exist(db_name):
            engine = sqlal.create_engine(f'mysql+mysqldb://root:비밀번호@127.0.0.1:3306/{db_name}', encoding='utf-8')
            print(f"{db_name} 데이터베이스와 연결되었습니다.")
            return engine.connect()

    def _if_exist(self, db_name):
        if db_name in self.DB_list:
            return True
        else:
            print(f"{db_name}는 존재하지 않는 데이터베이스입니다.")
            return False

    def _make_db(self, db_name):
        connection = self.engine.connect()
        connection.execute(f'CREATE DATABASE {db_name}')
        print(f"데이터베이스가 생성되었습니다.")

if __name__ == "__main__":
    check_db()._make_db('test')

프로그램 실행 결과, test 데이터베이스가 정상적으로 생성되었음을 확인할 수 있다.

더 나아가, 데이터베이스의 존재 여부를 판단할 수 있는 함수인 `def _if_exists(self, db_name):` 함수를 활용하여 해당 데이터베이스가 존재하지 않을 경우에만 데이터베이스를 생성하도록 할 수도 있다.
※ Line: 3, 10 ~ 13

    def _if_exist(self, db_name):
        if db_name in self.DB_list:
            print(f"존재하는 데이터베이스입니다.")
            return True
        else:
            print(f"{db_name}는 존재하지 않는 데이터베이스입니다.")
            return False

    def _make_db(self, db_name):
        if not self._if_exist(db_name):
            connection = self.engine.connect()
            connection.execute(f'CREATE DATABASE {db_name}')
            print(f"데이터베이스가 생성되었습니다.")

▶ 실행 결과 확인하기(이미 생성되어 있는 상태임. 생성하는 과정을 보고싶다면 MySQL에서 test 데이터베이스 삭제)

존재하는 데이터베이스입니다.

 

파이썬의 Magic Method 활용하기

위의 게시글을 읽어보고 왔다면 프로그램을 구현할 때 자원을 사용한 후에 반환하는 작업이 반드시 필요하다는 것을 알 수 있을 것이다. 물론 사용과 반환을 처리하는 과정에 있어서는 다양한 방법들이 있겠지만, 가장 대표적으로 사용되는 함수는 `with` 구문이다. 이 구문을 사용하는 방법에 관해서는 수없이 많은 게시글에서 설명하고 있으니 이는 제외하도록 하고(물론 아래에서 코드를 작성하는 방법은 살펴볼 예정), 여기서는 `with` 구문을 사용했을 때 함께 활용할 수 있는 Magic Method에 대해 보다 자세히 살펴보고자 한다.

사실 거창하게 설명한 것 치고는 별 거 없는 것처럼 느껴질 수도 있다. 왜냐하면 이전까지 클래스를 제작할 때 모든 클래스에 넣어주었던 함수이자 초기화 함수라 설명했던 `def __init__(self):` 함수도 Magic Method의 일부이기 때문이다. 이번에는 Magic Method의 종류에 대해 살펴보고자 한다. 

• `def __init__(self):` : 클래스가 호출되었을 때 가장 먼저 호출되는 함수(메소드)
• `def __enter__(self):` : 위의 `with` 구문에 진입하는 시점에 자동으로 호출되는 함수(메소드)
• `def __exit__(self, type, value, traceback):` : `with` 구문에서 나오기 직전에 자동으로 호출되는 함수(메소드)

그렇다면 이 내용을 가지고 sqlalchemy에 적용하기 위해서는 어떠한 방식으로 사용해야 할까 ? 일단 `with` 구문과 함께 작동할 `class sqlalCon:` 이라는 이름의 클래스를 기존에 생성해두었던 클래스의 아래에 생성해주도록 하자. 이 클래스는 MySQL과의 연결과 더불어 통신 과정에서 발생하는 `__enter__`와 `__exit__`의 처리를 담당할 것이다. 

## con_mysql.py ##
class sqlalCon:
    """Sqlalchemy Database Connection"""
    def __init__(self, db_name):
        pass
        
    def __enter__(self):
        pass
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

코드를 작성하기 전에 앞서, 위의 클래스에 포함되어 있는 `__enter__`나 `__exit__`같은 경우에는 `sqlalchemy` 라이브러리에서 제공하고 있으며 우리가 앞서 생성했던 `engine`을 사용할 때가 아니라, `session`을 사용할 때 보다 유용하게 활용할 수 있는 내용이긴 하다. 다만 `session`은 `engine`과 어느 정도 비슷한 개념이 아니라 보다 더 높은 수준에서 데이터베이스를 관리할 수 있는 방식이고 다루기 너무 복잡하니 여기서는 이 클래스의 기본 구조만 만들어놓고 넘어가도록 하자.

class sqlalCon:
    """Sqlalchemy Database Connection"""
    def __init__(self, db_name):
        print(f"sqlalCon 클래스가 호출되었습니다.")
        self.db_name = db_name
        self.session = None

    def __enter__(self):
        print(f"with 구문이 호출되었습니다.")
        self.engine = sqlal.create_engine(f'mysql+pymysql://root:a9985623@127.0.0.1:3306/{self.db_name}', encoding='utf-8', echo=False)
        Session = sqlalorm.sessionmaker(autocommit=False)
        self.session = Session(bind=self.engine)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"with 구문이 종료되었습니다.")
        self.session.close()
        print(f"세션을 종료하였습니다.")

실질적으로 `class sqlalCon:` 클래스는 초기화 함수에 정의되어 있는 `self.session`에 대한 내용만 봐도 알 수 있듯이 기본적으로 `engine`이 아닌 `session`을 활용하여 데이터를 처리할 때 보다 효과적이고 유용하게 사용할 수 있는 클래스이다. 하지만 우리가 차트 데이터를 저장할 때 사용할 판다스(`pandas`) 라이브러리의 `to_sql` 함수(메서드)는 기본적으로 `engine` 만을 사용하기 때문에 이 클래스는 유용하지 않을 수 있다. 따라서 `session`을 활용하여 데이터베이스를 관리하고 싶다면 따로 공부하고 적용해보길 권한다.

 

---

반응형

728x90

---

 

차트 데이터를 저장할 클래스 생성하기

앞의 단계까지 `class sqlalCon:` 클래스가 생성되었다면 이제 `with` 구문을 통해 이 클래스를 호출해서 사용하면 된다. 이제 이 기능을 수행할 클래스를 `class manage_db:`라는 이름의 클래스를 생성해주도록 하자. 그 후에 `def _save_table()`라는 이름으로 차트 데이터의 저장만을 담당할 함수를 생성해주도록 하자. 이 함수가 전달받아야 할 인자는 아래와 같다.

• `db_name` : 데이터베이스와 연결할 때 필요한 데이터베이스 이름('test' 등)
• `table_name` : 전달받은 차트 데이터(아래의 `table_data`)를 저장할 때 입력할 테이블 이름(A005930 등)
• `table_data` : 저장할 차트 데이터(차트 데이터를 조회한 후에 생성된 데이터프레임)

이제 `with` 구문과 함께 클래스에 `db_name` 변수를 전달해주되 그에 대한 인스턴스를 con이라는 변수에 저장하여 사용하면 되는데, 아래의 코드만 봐도 어렵지 않게 파악할 수 있겠지만 `try`, `except`, `finally`라는 세 가지 단계로 구분되어 있다는 걸 알 수 있다. 각각의 단계는 영단어를 해석한 그대로 보면 '시도하다', '예외적으로', '마지막으로'라는 의미를 갖고 있다. 다시 말해 `try` 구문 아래에서는 '실행시킬 코드'를 넣어주면 되고 `except` 구문 아래에서는 try 구문 아래에 입력해두었던 코드를 실행하는 과정 중에 '오류가 발생했을 경우에 실행시킬 코드'를 넣어주면 된다. 마지막으로 `finally` 구문 아래에는 앞선 단계를 모두 거친 후에, 오류가 발생했든지 발생하지 않았든지 간에 상관없이 '최종적으로 실행시킬 코드'를 넣어주면 된다. 이 단계에서는 `commit()` 함수를 통해 관련 데이터를 저장하는 작업을 수행하도록 하는 코드를 넣어놓긴 했지만 앞서 살펴봤듯이 지금까지의 코드는 `session`이 관여하는 부분이 없기 때문에 실질적으로는 아무런 기능도 수행하지 않을 것이다. 마지막으로 `with` 구문을 빠져나간 후의 단계에서는 MySQL과의 연결을 종료하면 되기 때문에 `dispose()`를 사용하여 `engine`을 종료시켜주도록 하자.

## con_mysql.py ##
class manage_db:
    def __init__(self):
        pass

    def _save_table(self, db_name, table_name, table_data):
        """
        table_data(데이터프레임, 차트 데이터 등)을 데이터베이스(db_name)에
        table_name(보통 종목코드가 됨)라는 이름으로 하여 저장"""
        with sqlalCon(f'{db_name}') as con:
            try:
                pass
            except Exception as e:
                con.session.rollback()
                print(f"오류가 발생했습니다. 데이터를 저장하지 않습니다. 오류 메시지:{e}")
            finally:
                con.session.commit()
                print(f"데이터를 저장하였습니다.")
        print(f"MySQL 연결을 종료합니다.")
        con.engine.dispose()    ## MySQL 프로그램 연결 종료
        print(f"MySQL과의 연결이 종료되었습니다.")

이제 차트 데이터를 MySQL에 저장하기 위해 마지막으로 살펴볼 내용은 판다스(`pandas`) 라이브러리에서 제공하는 `to_sql` 함수이다. 이 함수 역시 함수 이름을 그대로 해석해보면 그 기능을 알 수 있듯이, 데이터프레임을 MySQL로 저장하고자 할 때 사용하는 함수이다. 이 함수를 사용할 때 파라미터로 전달해주어야 하는 인자는 아래와 같다.

• `name`: 앞서 인자로 전달받은 저장할 테이블 이름(`table_name`)
• `con`: `sqlalchemy`를 통해 얻은 `engine`

그렇다면 이제 판다스 모듈을 활용하여 데이터프레임을 MySQL에 저장해보도록 하자. 아래의 조건문에 있는 `df`는 `a`를 기반으로 데이터프레임으로 만든 변수로 단순 예시 데이터에 해당하니 너무 개의치 않도록 하자. 아래의 코드를 실행시킨 후 MySQL WorkBench를 열어보면 test 데이터베이스 내에 hold_account라는 이름의 테이블이 저장되었음을 확인할 수 있다.
※ Line: 16, 17

## con_mysql.py ##
class manage_db:
    def __init__(self):
        pass

    def _save_table(self, db_name, table_name, table_data):
        """
        table_data(데이터프레임, 차트 데이터 등)을 데이터베이스(db_name)에
        table_name(보통 종목코드가 됨)라는 이름으로 하여 저장"""
        if type(table_data) != pd.DataFrame:    ## 데이터타입이 데이터프레임이 아닌 경우
            print(f"전달받은 데이터의 데이터타입이 데이터프레임이 아닙니다.")
            return False                        ## False를 반환하고 종료

        with sqlalCon(f'{db_name}') as con:
            try:
                table_data.to_sql(name=table_name, con=con.engine, index=False)
                print(f"데이터가 저장되었습니다. (테이블명:{table_name})")
            except Exception as e:
                con.session.rollback()
                print(f"오류가 발생했습니다. 데이터를 저장하지 않습니다. 오류 메시지:{e}")
            finally:
                con.session.commit()
                print(f"데이터를 저장하였습니다.")
        print(f"MySQL 연결을 종료합니다.")
        con.engine.dispose()    ## MySQL 프로그램 연결 종료
        print(f"MySQL과의 연결이 종료되었습니다.")

if __name__ == "__main__":
    a = {'item_code':['005930', '000020'], 'buy_date':['20230101', '20240101']}
    df = pd.DataFrame(a, columns=['item_code', 'buy_date'])
    manage_db()._save_table('test', 'hold_account', df)

▶ 실행 결과 확인하기(테이블 생성 여부는 MySQL WorkBench에서 참고)

sqlalCon 클래스가 호출되었습니다.
with 구문이 호출되었습니다.
데이터가 저장되었습니다. (테이블명:hold_account)
데이터를 저장하였습니다.
with 구문이 종료되었습니다.
세션을 종료하였습니다.
MySQL 연결을 종료합니다.
MySQL과의 연결이 종료되었습니다.

 

 

---