SQLite 활용하는 예시 (with Python)

개인적으로 뭘 해보고자 하는게 있어서 요즘 잠깐 잠깐 만들고 있는 프로젝트가 있다.

- NAVER API를 이용해서 블로그 검색하기 (with Python)

 

그런데, 검색된 블로그 포스팅들을 화면에 출력만 할 것이 아니라 저장을 해놓고 싶은데,

그냥 JSON 형태로 저장하거나 CSV 형태로 저장하기엔 싫어서 SQLite를 이용해보려고 한다.

 

SQLite에 대한 기본적인 사항은 다음 포스팅을 참고하기 바란다.

- SQLite에 대해서 알아보자

 

 

0. 개발 환경

- OS: Ubuntu 20.04

- Lang: Python 3.10.9

 

 

1. 기본 코드

블로그 포스팅을 검색하는 코드는 이미 아래와 같이 살펴봤다.

- NAVER API를 이용해서 블로그 검색하기 (with Python)

 

 

2. SQLite 활용

기본 작성한 코드에서 이것 저것 좀 많이 업그레이드 시켰다.

 

에러메시지는 log로 남기도록 했고,

태그를 지우는 것은 BeautifulSoup 패키지를 활용하도록 했다.

 

import requests
import os
import sqlite3
from bs4 import BeautifulSoup
import logging

# 로깅 설정
logging.basicConfig(level=logging.ERROR)

class BlogPost:
    def __init__(self, title, description, link, postdate):
        self.title = self.clean_html(title)
        self.description = self.clean_html(description)
        self.link = link
        self.postdate = postdate

    def clean_html(self, text):
        if text:
            return BeautifulSoup(text, "html.parser").get_text()
        return ""

    def __str__(self):
        return f"Title: {self.title}\nDescription: {self.description}\nURL: {self.link}\nPost Date: {self.postdate}\n"

class NaverBlogSearcher:
    def __init__(self):
        self.client_id = os.getenv("NAVER_CLIENT_ID")  # 환경변수에서 Client ID 읽어오기
        self.client_secret = os.getenv("NAVER_CLIENT_SECRET")  # 환경변수에서 Client Secret 읽어오기

        # 환경변수 값이 없을 때 예외 처리
        if not self.client_id or not self.client_secret:
            logging.error("NAVER_CLIENT_ID and NAVER_CLIENT_SECRET must be set as environment variables.")
            raise ValueError("NAVER_CLIENT_ID and NAVER_CLIENT_SECRET must be set as environment variables.")

    def get_blog_posts(self, query, display=10):
        # 요청 URL 및 헤더 구성
        url = f"https://openapi.naver.com/v1/search/blog.json?query={query}&display={display}&sort=date"
        headers = {
            "X-Naver-Client-Id": self.client_id,
            "X-Naver-Client-Secret": self.client_secret,
        }

        # 네이버 API에 요청 보내기
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            logging.error(f"Network error occurred: {e}")
            return []
        
        if response.status_code == 200:
            return self.parse_response(response.json())
        else:
            print(f"Error: {response.status_code}, {response.text}")
            return []

    def parse_response(self, data):
        # 응답 데이터 파싱
        blogs = data.get("items", [])
        blog_posts = [BlogPost(blog.get("title"), blog.get("description"), blog.get("link"), blog.get("postdate")) for blog in blogs]
        return blog_posts

class BlogPostDBHandler:
    def __init__(self, db_name="blog_posts.db"):
        self.db_name = db_name
        self.create_table()

    def create_table(self):
        # SQLite 데이터베이스 연결
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()

            # 테이블 생성
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS BlogPosts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT,
                    description TEXT,
                    link TEXT UNIQUE,
                    postdate TEXT
                )
            ''')

    def save_to_db(self, blog_posts):
        # SQLite 데이터베이스 연결
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()

            # 데이터 삽입 (중복 확인)
            for post in blog_posts:
                cursor.execute('''
                    SELECT COUNT(*) FROM BlogPosts WHERE link = ?
                ''', (post.link,))
                if cursor.fetchone()[0] == 0:
                    cursor.execute('''
                        INSERT INTO BlogPosts (title, description, link, postdate) 
                        VALUES (?, ?, ?, ?)
                    ''', (post.title, post.description, post.link, post.postdate))

    def get_all_posts(self):
        # SQLite 데이터베이스 연결
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()

            # 저장된 블로그 포스팅의 모든 필드 가져오기
            cursor.execute('''
                SELECT title, description, link, postdate FROM BlogPosts
            ''')
            rows = cursor.fetchall()

        # 필드명을 갖는 dict 형태로 변환
        posts = [
            {"title": row[0], "description": row[1], "link": row[2], "postdate": row[3]} 
            for row in rows
        ]
        return posts

if __name__ == "__main__":
    # "동탄" 검색어로 최신 블로그 정보 가져오기
    try:
        searcher = NaverBlogSearcher()
        blog_posts = searcher.get_blog_posts("동탄")
        
        # 검색 결과 출력
        # for post in blog_posts:
        #     print(post)
        
        # 검색 결과를 SQLite 데이터베이스에 저장
        db_handler = BlogPostDBHandler()
        db_handler.save_to_db(blog_posts)

        # 저장된 블로그 포스팅의 제목 출력
        posts = db_handler.get_all_posts()
        for post in posts:
            print(f"Title: {post['title']}")
    except ValueError as e:
        print(e)

 

TABLE이 존재하지 않으면 create 하도록 했으며,

블로그 포스팅을 insert 하기 전에 기존에 있는지 미리 확인한 후에 없을 때에만 insert 하도록 했다.

 

아직은 PoC 성격의 스크립트 이기에,

SQLite에 저장한 이후에 전체 포스팅 목록을 SQLite로부터 읽어와서 제목들을 출력하도록 했다.

 

그리고,

추가 설치해야할 패키지들이 있기에 requirements.txt 파일도 업데이트 되었다.

 

beautifulsoup4==4.12.3
requests==2.32.3

 

 

3. Execute

실행 결과도 확인 해보고, database 내용도 살펴보자.

 

> dotenvx run -f .env -- python main.py

 

 

이렇게 저장을 하면 database 파일이 하나 생성되는 것을 확인할 수 있다.

- blog_posts.db

 

 

 

4. Check

db 파일을 확인해보기 위해서 "DB Browser for SQLite"를 사용해보겠다.

 

 

설치는 그냥 패키지 설치하면 된다.

 

> sudo apt-get install sqlitebrowser

 

잘 설치되어 있는 것을 볼 수 있다.

 

 

실행하고, "데이터베이스 열기" 하고 db 파일을 선택해주면 된다.

 

 

"데이터 보기" 탭을 선택하면 저장된 내역을 볼 수 있다.

 

 

 

오늘은 여기까지~