SW정리: Sqlite3, python에서 DB사용, dict type db 저장 예제

SQLite3

SQLite는 별도의 서버 프로세스가 필요 없고 SQL 질의 언어의 비표준 변형을 사용하여 데이터베이스에 액세스할 수 있는 경량 디스크 기반 데이터베이스를 제공하는 라이브러리입니다. Python에서도 쉽게 사용이 가능합니다.

여기 예제에서는 dictinary type을 저장하는 예제로 만들어 보도록 하겠습니다.

설치

pip install pysqlite3

python 3.8

SQL 기본 문법

SQL의 기본 명령에 대해서는 알고 있어야 합니다. 아래 링크에서 확인이 가능합니다.

https://www.w3schools.com/sql/

들어가기에 앞서 sqlite에서 db는 파일 이름 단위로 생성이 되며, db안에는 table이 여러개 존재할 수 있으며, 하나의 table에는 여러개의 레코드가 존재 가능합니다.

기본적으로 CREATE TABLE(테이블 생성), INSERT INTO(레코드 추가), UPDATE(레코드 업데이트), SELECT(레코드 검색), DELETE FROM (레코드 삭제)정도는 알아야 합니다.

python 에서 SQL 기본 명령 사용하기

최초 db를 connect 해야 합니다.

import sqlite3
con = sqlite3.connect('example.db')

connect되면 connect object가 생성되면 여기에서 cursor() 메소드를 이용하여 아래와 같이 sql command 이용이 가능합니다. 아래 예제에서 cur.execute() 입니다.

cur = con.cursor()

# Create table
cur.execute('''CREATE TABLE stocks
               (date text, trans text, symbol text, qty real, price real)''')

# Insert a row of data
cur.execute("INSERT INTO stocks VALUES ('2006-01-05','BUY','RHAT',100,35.14)")

# Save (commit) the changes
con.commit()

# We can also close the connection if we are done with it.
# Just be sure any changes have been committed or they will be lost.
con.close()

db에 변경이 일어나면 마지막에 connect.commit() 메소드를 호출 하도록 합니다. commit전에 rollback이 일어나면 excute했던 동작들이 모두 취소됩니다.

dictionary 저장 하기 예제

dictionary를 저장하는것은 json으로 text형태로 쉽게 저장이 가능하지만 값이 빈번하게 저장이 될때 power fail이 발생하는 경우 손상될 가능성이 높습니다.

저장된 파일의 check sum을 기록한다던지 여러가지 방법이 있겠지만, 가장 간단하게 처리가 가능한 것이 db를 이용하여 처리하는 것이 가장 간단한 방법입니다.

예제코드

import os
import sqlite3
import json

# key must be str type
class sqlite_dict():
    def __init__(self, dbfilename):
        self.dbfilename = dbfilename
        self.table_name = "test"
        self.is_connect = False
        self.con = None

    def drop_db_file(self):
        if os.path.isfile(self.dbfilename):
            os.remove(self.dbfilename)
        self.is_connect = False

    def connect(self):
        self.con = sqlite3.connect(self.dbfilename)
        self.is_connect = True

    def close(self):
        if self.is_connect:
            self.con.close()
            self.is_connect = False
            self.con = None

    def set_table_name(self, table_name):
        self.table_name = table_name

    def create_table(self):
        cur = self.con.cursor()
        cur.execute(f"CREATE TABLE {self.table_name} ( dict_key text primary key, json_dump text)")
        self.con.commit()

    # dict를 하나 또는 여러개를 DB에 저장합니다.
    # dupkey_process
    # "ignore":에러가 안나는 것만 부분 업데이트
    # "forced":기존값 모두 업데이트
    # "error":하나가 에러나면 모두 업데이트 중지함
    # return 0 : 에러없음, -1:에러
    def save_dict(self,dict_data,dupkey_process="error"):
        cur = self.con.cursor()
        for key in dict_data:
            value = dict_data[key]
            json_dump = json.dumps(value)
            print(f"{key}//{json_dump}")
            try:
                cur.execute(f"INSERT INTO '{self.table_name}' VALUES ('{key}','{json_dump}')")
            except sqlite3.OperationalError as ex:
                if str(ex).startswith("no such table:"):
                    self.create_table()
                    cur.execute(f"INSERT INTO '{self.table_name}' VALUES ('{key}','{json_dump}')")
            except sqlite3.IntegrityError as ex:
                if str(ex).startswith("UNIQUE constraint failed:"):
                    if dupkey_process == "ignore":
                        pass
                    if dupkey_process == "error":
                        self.con.rollback()
                        return -1
                    if dupkey_process == "forced":
                        cur.execute(f"UPDATE '{self.table_name}' SET json_dump='{json_dump}' WHERE dict_key='{key}'")
        self.con.commit()
        return 0

    def get_data(self, find_key):
        if not self.is_connect:
            self.connect()
        cursor = self.con.cursor()
        try:
            cursor.execute(f"SELECT json_dump FROM '{self.table_name}' WHERE dict_key='{find_key}'")
        except sqlite3.OperationalError as ex:
            if str(ex).startswith("no such table:"):
                cursor.close()
                return None
        row = cursor.fetchone()
        cursor.close()
        if row is None:
            return None
        dict_data = json.loads(row[0])
        return dict_data

    def get_all_data(self):
        if not self.is_connect:
            self.connect()
        cursor = self.con.cursor()
        try:
            cursor.execute(f"SELECT * FROM '{self.table_name}'")
        except sqlite3.OperationalError as ex:
            if str(ex).startswith("no such table:"):
                cursor.close()
                return None
        rows = cursor.fetchall()
        cursor.close()
        if rows is None:
            return None
        ret = {}
        for row in rows:
            ret[row[0]] = json.loads(row[1])
        return ret

    def del_key(self, find_key):
        if not self.is_connect:
            self.connect()
        cursor = self.con.cursor()
        try:
            cursor.execute(f"DELETE FROM '{self.table_name}' WHERE dict_key='{find_key}'")
        except sqlite3.OperationalError as ex:
            if str(ex).startswith("no such table:"):
                cursor.close()
                return False
        cursor.close()
        self.con.commit()
        return True

if __name__ == "__main__":
    sd = sqlite_dict("test.db")
    sd.connect()
    sd.set_table_name("test_table")
    print("** add item")
    sd.save_dict({"abc" : [1, 2, 3, 4, 5,"abc"]})
    print("** add items 2")
    sd.save_dict({"abc" : [1, 2, "한글"], "kbz" : 2} , dupkey_process="forced")
    print("** alldata")
    print(sd.get_all_data())
    print("** read item")
    print(sd.get_data("abc"))
    print(sd.get_data("kbz"))
    print(sd.get_data("kcc"))
    print("** del item")
    print(sd.del_key("kca"))
    print(sd.del_key("kbz"))
    print("** alldata")
    print(sd.get_all_data())
    sd.close()

실행화면

** add item
abc//[1, 2, 3, 4, 5, "abc"]
** add items 2
abc//[1, 2, "\ud55c\uae00"]
kbz//2
** alldata
{'abc': [1, 2, '한글'], 'kbz': 2}
** read item
[1, 2, '한글']
2
None
** del item
True
True
** alldata
{'abc': [1, 2, '한글']}

bug가 발견되어 변환 코드를 추가해야 합니다. 아래 링크 참고

https://swlock.blogspot.com/2022/10/sqlite3-unrecognized-token.html

SW정리

2022년 7월 10일 일요일

Sqlite3, python에서 DB사용, dict type db 저장 예제