Flask에 SQLite 연동하기

개발일지

Flask에 SQLite 연동하기

B1001101 2022. 5. 23. 23:32

예전에 채용공고를 크롤링하는 JobScrapper라는 사이트를 만들었었다. 기존에는 검색 결과를 임시로 딕셔너리에 저장하는 방식을 사용했는데, 이를 SQLite에 저장하는 방식으로 바꿔보기로 했다.

데이터 모델링

우선 ERDCloud를 사용해서 다음과 같이 데이터베이스를 설계했다.

Keywords
- 검색된 키워드들을 저장
- 키워드는 중복되지 않는 고유한 값이므로 Primary key로 설정
Results
- 검색 결과를 저장
- 자동으로 아이디 부여
- 키워드를 Foreign Key로 설정 (Keywords 테이블에서 참조)

SQLite 사용법

SQLite의 자료형은 5가지이다. 동적 자료형 시스템을 사용하고 있기 때문에, 특정 타입의 칼럼을 선언했더라도 해당 칼럼에는 모든 종류의 자료를 저장할 수 있다.
- NULL: 빈 정보, 알 수 없는 정보
- INTEGER: 정수(양수 또는 음수), 가변 크기를 가질 수 있음
- REAL: 10진수 실수(8바이트 부동 소수점 사용)
- TEXT: 문자, 길이 제한 없음, 다양한 인코딩 지원
- BLOB: 모든 종류의 데이터를 저장할 수 있는 대형 바이너리 오브젝트
참조

SQLite 연동

Flask에서 SQLite 사용하기
- DB 생성 / 연결: conn = sqlite3.connect("db 경로")
- 커서 설정: cur = conn.cursor()
- 쿼리 실행: cur.execute("SQL문")
  - 파라미터: SQL에서 파라미터가 들어갈 위치를 ?로 표시, 값은 튜플 형태로 전달
    cur.execute("SQL문 = ?", (param1, param2, ...))
- 결과 받아오기: rows = cur.fetchall()
- 커밋: conn.commit()
- 롤백: conn.rollback()
- DB 연결 종료: conn.close()
INSERT, DELETE와 같이 데이터에 변화가 발생하는 쿼리들은 try~except~finally문으로 예외를 처리하였다. 정상적으로 실행이 완료되었을 경우에는 commit, 에러가 발생했을 경우에는 rollback을 하도록 했다.

사용하기 편리하도록 필요한 기능별로 함수를 작성하였다. 결과는 기존 데이터 형식과 동일한 형태로 변환하여 리턴하도록 했다.
- create_db(): 데이터베이스, 테이블 생성
- get_keywords(): 키워드 리스트
- get_results(): 키워드별 검색결과 리스트
- add_results(): 검색결과 추가
- delete_results(): 검색결과 삭제
참조

db.py

import sqlite3
import sys

def create_db():
    # 데이터베이스 생성
    conn = sqlite3.connect("database.db")

    # 테이블 생성
    conn.execute('''
        CREATE TABLE IF NOT EXISTS KEYWORDS(
            KEYWORD TEXT PRIMARY KEY,
            LASTSEARCHED TEXT DEFAULT CURRENT_TIMESTAMP
        );
    ''')
    conn.execute('''
        CREATE TABLE IF NOT EXISTS RESULTS(
            ID INTEGER PRIMARY KEY AUTOINCREMENT,
            KEYWORD TEXT,
            TITLE TEXT,
            COMPANY TEXT,
            LOCATION TEXT,
            TIME TEXT,
            LINK TEXT,
            FOREIGN KEY(KEYWORD) REFERENCES KEYWORDS
        );
    ''')

    conn.commit()
    conn.close()


def get_keywords():
    conn = sqlite3.connect("database.db")
    cur = conn.cursor()
    cur.execute("SELECT KEYWORD FROM KEYWORDS;")
    rows = cur.fetchall()
    cur.close()
    return [r[0] for r in rows]


def get_results(keyword):
    conn = sqlite3.connect("database.db")
    cur = conn.cursor()
    cur.execute("SELECT * FROM RESULTS WHERE KEYWORD = ?;", (keyword,))
    rows = cur.fetchall()
    res = []
    for r in rows:
        res.append({
            "title": r[2],
            "company": r[3],
            "location": r[4],
            "time": r[5],
            "link": r[6]
        })
    cur.close()
    return res


def add_result(keyword, jobs):
    try:
        conn = sqlite3.connect("database.db")
        cur = conn.cursor()
        cur.execute("INSERT INTO KEYWORDS(KEYWORD) VALUES(?);", (keyword,))
        for job in jobs:
            cur.execute('''
                    INSERT INTO RESULTS(
                        KEYWORD, 
                        TITLE,
                        COMPANY,
                        LOCATION,
                        TIME,
                        LINK
                    ) VALUES(?,?,?,?,?,?);
                ''', 
                (keyword, 
                job["title"], 
                job["company"], 
                job["location"],
                job["time"],
                job["link"])
            )
        conn.commit()
    except:
        conn.rollback()
    finally:
        conn.close()


def delete_result(keyword):
    try:
        conn = sqlite3.connect("database.db")
        cur = conn.cursor()
        cur.execute("DELETE FROM KEYWORDS WHERE KEYWORD = ?;", (keyword,))
        cur.execute("DELETE FROM RESULTS WHERE KEYWORD = ?", (keyword,))
        conn.commit()
    except:
        conn.rollback()
    finally:
        conn.close()

저작자표시