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

앞에서 sqlite를 이용해서 python dict type을 저장하는 것을 만들어 봤는데...

https://swlock.blogspot.com/2022/07/sqlite3-python-db-dict-type-db.html

실제 프로젝트에선 몇 가지 기능이 더 필요해서 업그레이드 하였습니다.

추가로 필요했던 기능은 전체 레코드 갯수 와 오래된 레코드 삭제하는 기능입니다.

단순 db에 저장만 하는 것이 아닌 cache 형태로 관리하려면 일정한 갯수를 유지하기 위한 방법이 필요했습니다.

그래서 table에 timestamp정보를 넣고 오래된 항목을 삭제하는 기능을 넣었습니다.

아래 내용을 보기전에 기존 구현된 내용을 위 링크를 이용하여 먼저 읽기를 바랍니다.

table에 신규 timestamp 컬럼 추가

이건 간단합니다. timestamp REAL 타입을 추가했습니다.

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

저장시 timestamp 추가

이것도 values 항목에 첫번째로 넣어주면 되고, datetime.now()의 timestamp() 메소드를 이용하였습니다.

def save_dict(self, dict_data, dupkey_process="error"):
    nowtime = datetime.datetime.now().timestamp()
    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 ('{nowtime}','{key}','{json_dump}')")

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

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

timestamp 읽기

data를 읽어올때 timestamp도 같이 읽어 오도록 하였습니다.

sql select 명령에 timestamp를 붙였습니다.

cursor.execute(f"SELECT json_dump,timestamp FROM '{self.table_name}' WHERE dict_key='{find_key}'")

전체 레코드 갯수

sql의 COUNT를 이용하였습니다. 레코드의 전체 카운트를 리턴합니다.

# 전체 카운트 리턴
def get_count_all(self):
    if not self.is_connect:
        self.connect()
    cursor = self.con.cursor()
    result = 0
    try:
        cursor.execute(f"SELECT COUNT(*) FROM '{self.table_name}'")
        result = cursor.fetchone()[0]
    except sqlite3.OperationalError as ex:
        if str(ex).startswith("no such table:"):
            cursor.close()
            return -1
    cursor.close()
    return result

오래된 레코드 가져오기

timestamp가 갱신할 때마다 저장되므로 order by timestamp 를 이용하여 마지막 레코드를 읽어오도록 구현하였습니다. 

# 가장 오래된 레코드 가져오기
# dict, timestamp
def get_old_record_one(self):
    if not self.is_connect:
        self.connect()
    cursor = self.con.cursor()
    try:
        cursor.execute(f"SELECT * FROM '{self.table_name}' ORDER BY timestamp ASC LIMIT 1")
    except sqlite3.OperationalError as ex:
        if str(ex).startswith("no such table:"):
            cursor.close()
            return None, None
    row = cursor.fetchone()
    print(row)
    cursor.close()
    if row is None:
        return None, None
    ret = {row[1]: json.loads(row[2])}
    return ret, row[0]

특정 시간 이전 레코드 삭제

구현 방법은 where 조건절에 timestamp정보를 넣어서 레코드를 얻어오고, 해당 레코드에서 키를 이용하여 기존 존재하는 del_key() 메소드로 삭제를 하도록 구현하였습니다.

# 특정 시간 이전 레코드 삭제하기
# -60 : -60 초 이전 레코드 삭제
def del_old_record(self, time_old_sec):
    nowtime = datetime.datetime.now().timestamp()
    print(nowtime+time_old_sec)
    if not self.is_connect:
        self.connect()
    cursor = self.con.cursor()
    try:
        cursor.execute(f"SELECT * FROM '{self.table_name}' WHERE timestamp < {nowtime+time_old_sec}")
    except sqlite3.OperationalError as ex:
        if str(ex).startswith("no such table:"):
            cursor.close()
            return -1
    rows = cursor.fetchall()
    #print(rows)
    cursor.close()
    count = 0
    if rows is not None:
        for row in rows:
            self.del_key(row[1])
            count = count + 1
    return count

python에서는 dict type을 정말 많이 이용하게 되고, 이것을 쉽게 sqlite를 이용하여 db를 저장하도록 구현한 내용입니다. 

sourcecode/dict_sqlite.py at main · donarts/sourcecode · GitHub