5. 구현 - 데이터 접근 영역(DB)

1. 목표

 

도서관리 프로그램과 MySQL(DB)간에 통신하기 위한 "DatabaseManager" 클래스를 구현합니다.

 

DatabaseManager 클래스 구현 목적은

UI와 DB 사이에 "중간 계층(DatabaseManager)"을 두고, 모든 SQL 실행/조회 흐름을 표준화 하고

*UI 코드에서 DB 코드를 분리(SRP) 합니다.

 

그리고 다음 3가지를 안정적으로 제공합니다.

• DB 연결(connect)
• 데이터 변경 쿼리 실행(execute_query: INSERT/UPDATE/DELETE)
• 데이터 조회(fetch_data: SELECT)

구현 후에는 DB 연결, 실행, 조회를 하나의 공통 클래스(DatabaseManager)로 분리하여 "재사용" 관점에서 이해합니다.
또한 UI(PySide6)가 없어도 DatabaseManager 단독으로 DB 조회/실행이 가능함도 확인합니다.

2. 전체 로직

(1) DB 테이블 생성

-- 1. 데이터베이스 선택/생성
CREATE DATABASE IF NOT EXISTS library_db;
USE library_db;
 
-- 2. 기존 테이블이 있다면 안전하게 삭제 (옵션: 테스트 환경에서만 사용)
-- DROP TABLE IF EXISTS issues;
-- DROP TABLE IF EXISTS members;
-- DROP TABLE IF EXISTS books;
 
-- 3. 도서 테이블 (book_id 사용)
CREATE TABLE IF NOT EXISTS books (
    book_id VARCHAR(50) PRIMARY KEY,
    title VARCHAR(100),
    author VARCHAR(100),
    publisher VARCHAR(100),
    is_available BOOLEAN DEFAULT TRUE
);
 
-- 4. 회원 테이블 (member_id 사용)
CREATE TABLE IF NOT EXISTS members (
    member_id VARCHAR(50) PRIMARY KEY,
    name VARCHAR(50),
    mobile VARCHAR(20),
    email VARCHAR(100)
);
 
-- 5. 대출 기록 테이블
CREATE TABLE IF NOT EXISTS issues (
    issue_id INT AUTO_INCREMENT PRIMARY KEY,
    book_id VARCHAR(50),
    member_id VARCHAR(50),
    issue_date DATETIME DEFAULT CURRENT_TIMESTAMP,
    renew_count INT DEFAULT 0,
    FOREIGN KEY (book_id) REFERENCES books(book_id) ON DELETE CASCADE,
    FOREIGN KEY (member_id) REFERENCES members(member_id) ON DELETE CASCADE
);
 
 
-- 6. 샘플 데이터 입력
INSERT INTO members(member_id, name, mobile, email) VALUES
('M001', '김코딩', '010-1111-2222', 'kim@demo.com'),
('M002', '이파이', '010-3333-4444', 'lee@demo.com');
 
INSERT INTO books(book_id, title, author, publisher, is_available) VALUES
('B001', '파이썬 기초', '홍길동', '코딩출판', TRUE),
('B002', 'PySide6 실습', '김개발', 'Qt출판', TRUE),
('B003', '데이터베이스 입문', '박DB', 'SQL출판', TRUE);

 

 

(2) config.py

# config.py

DB_CONFIG = {
    "host": "localhost",
    "user": "root",
    "password": "1234",
    "database": "library_db",
}

 

 

(3) database_manager.py 클래스 구현

# database_manager.py

import mysql.connector as mc
from config import DB_CONFIG


class DatabaseManager:
    def __init__(self, config=None):
        self.config = config or DB_CONFIG
        self.conn = None

    def connect(self):
        try:
            self.conn = mc.connect(**self.config)
            return self.conn
        except Exception as e:
            print(f"[DB 연결 오류] {e}")
            return None

    def execute_query(self, query, params=None):
        conn = self.connect()
        if conn is None:
            return False
        try:
            cursor = conn.cursor()
            cursor.execute(query, params)
            conn.commit()
            return True
        except Exception as e:
            print(f"[쿼리 실행 오류] {e}")
            return False
        finally:
            if conn:
                conn.close()
                    
    def fetch_data(self, query, params=None):
        conn = self.connect()
        if conn is None:
            return []
        try:
            cursor = conn.cursor()
            cursor.execute(query, params)
            return cursor.fetchall()
        except Exception as e:
            print(f"[데이터 조회 오류] {e}")
            return []
        finally:
            if conn:
                conn.close()

 

 

(4) 테스트용 GUI 임시 사용

# GUI_Test.py

from PySide6.QtGui import Qt
from PySide6.QtWidgets import QMainWindow, QWidget, QVBoxLayout, QPushButton, QHBoxLayout, QTableWidget, QLabel, \
    QHeaderView, QMessageBox, QTableWidgetItem

from database_manager import DatabaseManager
from config import DB_CONFIG
db = DatabaseManager(DB_CONFIG)

class DbTestWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("DB 로직 테스트 (4단계)")
        self.resize(900, 600)

        root = QWidget()
        self.setCentralWidget(root)
        layout = QVBoxLayout(root)
        layout.setContentsMargins(20, 20, 20, 20)
        layout.setSpacing(10)

        top = QHBoxLayout()
        self.btn_ping = QPushButton("연결 테스트")
        self.btn_members = QPushButton("members 조회")
        self.btn_books = QPushButton("books 조회")
        top.addWidget(self.btn_ping)
        top.addWidget(self.btn_members)
        top.addWidget(self.btn_books)
        top.addStretch()
        layout.addLayout(top)

        self.status = QLabel("상태: 대기")
        self.status.setAlignment(Qt.AlignmentFlag.AlignLeft)
        layout.addWidget(self.status)

        self.table = QTableWidget()
        self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
        layout.addWidget(self.table)

        self.btn_ping.clicked.connect(self.ping_db)
        self.btn_members.clicked.connect(self.load_members)
        self.btn_books.clicked.connect(self.load_books)

    def ping_db(self):
        conn = db.connect()
        if conn is None:
            QMessageBox.warning(self, "실패", "DB 연결 실패 (콘솔 로그 확인)")
            self.status.setText("상태: 연결 실패")
            return
        conn.close()
        QMessageBox.information(self, "성공", "DB 연결 성공")
        self.status.setText("상태: 연결 성공")

    def set_headers(self, headers):
        self.table.setRowCount(0)
        self.table.setColumnCount(len(headers))
        self.table.setHorizontalHeaderLabels(headers)

    def fill_rows(self, rows):
        self.table.setRowCount(0)
        for r, row_data in enumerate(rows):
            self.table.insertRow(r)
            for c, data in enumerate(row_data):
                self.table.setItem(r, c, QTableWidgetItem(str(data)))

    def load_members(self):
        self.set_headers(["member_id", "name", "mobile", "email"])
        rows = db.fetch_data("SELECT member_id, name, mobile, email FROM members ORDER BY member_id")
        self.fill_rows(rows)
        self.status.setText(f"상태: members {len(rows)}건 조회")

    def load_books(self):
        self.set_headers(["book_id", "title", "author", "publisher", "is_available"])
        rows = db.fetch_data("SELECT book_id, title, author, publisher, is_available FROM books ORDER BY book_id")
        self.fill_rows(rows)
        self.status.setText(f"상태: books {len(rows)}건 조회")

 

 

(5) main.py

# main.py

import sys

from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication

from GUI_Test import DbTestWindow

if __name__ == "__main__":
    app = QApplication(sys.argv)
    font = QFont("Malgun Gothic", 12)
    app.setFont(font)

    window = DbTestWindow()
    window.show()
    sys.exit(app.exec())

 

  

(6) 현재 학습 단계 실행 구조

 

프로그램은 DB에 직접 접속하지 않고, 항상 DatabaseManager를 통해서만 DB를 사용합니다.

DB 테이블이 이미 만들어져 있다는 전제에서, 흐름은 다음과 같습니다.

 

1. GUI
   • 페이지(예: DtTestWindow)가 버튼 클릭 이벤트로 기능 실행
2. DatabaseManager
   1. connect()로 DB 연결
   2. cursor로 SQL 실행
   3. 변경 쿼리면 commit(), 조회면 fetchall()
   4. finally에서 연결 close()로 자원 정리
3. GUI 
   1. DB 결과 반영

---

 

ㄴ 2.1 import + 주석

(1) 필수 패키지 설치하기

pip install PySide6 mysql-connector-python

 

 

(2) mysql.connector

• • MySQL DB에 접속하기 위한 라이브러리입니다.
  • MySQL 접속(connect), 커서(cursor), SQL 실행(execute), 결과 조회(fetchall), 커밋(commit)

 

ㄴ 2.2 GUI + 주석

이 단계의 본 목적은 DatabaseManager이므로, 실제 프로젝트 UI는 만들지 않습니다.
대신 DB 연결과 조회를 검증하기 위한 "테스트 GUI"를 임시로 구성합니다.
테스트 GUI는 2번 항목을 참고합니다.

 

ㄴ 2.3 Signal & Slot + 주석

DatabaseManager 자체는 QWidget이 아니므로 시그널/슬롯이 없습니다.
시그널/슬롯은 테스트 GUI에서만 사용합니다.
테스트 GUI는 2번 항목을 참고합니다.

 

ㄴ 2.4. <DatabaseManager> 요약

(1) DB_CONFIG: 전역 설정 영역

• host/user/password/database 같은 접속 정보를 한 곳에 모읍니다.
• 이후 단계에서 UI가 늘어나도 DB 설정은 동일한 위치에서 관리됩니다

 

 

(2) DatabaseManager: DB 접근(중간 계층)

• UI는 SQL을 직접 실행하지 않고, db.execute_query(), db.fetch_data()만 호출합니다.
• 결과적으로 페이지 코드(회원/도서/대출)가 깔끔해지고 테스트가 쉬워집니다.
• 설정 → 연결 → 실행 준비 → 쿼리 실행 → 결과 처리 → 종료

# DatabaseManager 핵심 

# 0) 설정 준비 DB_CONFIG: 전역 설정 영역
config = {
    "host": "...",
    "user": "...",
    "password": "...",
    "database": "...",
}

# 1) 연결 생성
conn = mc.connect(**config)

# 2) 실행 준비
cursor = conn.cursor()

# 3) 쿼리 실행
cursor.execute(sql, params)

# 4) 결과 처리
conn.commit()            # INSERT/UPDATE/DELETE
rows = cursor.fetchall() # SELECT

# 5) 종료(필수)
conn.close()

 

 

ㄴ 2.5. <DatabaseManager> 상세 주석

(1) config.py

# config.py
# 학습용, DB 접속 정보 예시 예시
# 학습 이후, DB 접속 정보 공유 시 비밀번호는 반드시 분리(환경변수/.env) 권장

DB_CONFIG = {
    "host": "localhost",
    "user": "root",
    "password": "1234",
    "database": "library_db",
}

• DB_CONFIG를 분리해두면 main 코드가 깔끔해지고 재사용이 쉬워집니다.
• 실제 GitHub 업로드 시에는 반드시 .env 또는 환경변수로 분리해야 합니다.

 

(2) database_manager.py

# database_manager.py
# 3단계: DatabaseManager 단독 분석용 핵심 클래스

import mysql.connector as mc
from config import DB_CONFIG


class DatabaseManager:
    """
    DatabaseManager 역할
    - connect(): DB 연결 생성
    - execute_query(): INSERT/UPDATE/DELETE 실행 후 True/False 반환
    - fetch_data(): SELECT 실행 후 list[tuple] 반환
    """

    def __init__(self, config=None):
        # config를 외부에서 주입할 수 있게 설계(재사용 관점)
        self.config = config or DB_CONFIG
        self.conn = None

    def connect(self):
        """
        매 호출마다 새 연결을 생성하는 구조(단순한 대신 성능 비용 존재)
        - 성공: connection 반환
        - 실패: None 반환
        """
        try:
            self.conn = mc.connect(**self.config)
            return self.conn
        except Exception as e:
            print(f"[DB 연결 오류] {e}")
            return None

    def execute_query(self, query, params=None):
        """
        INSERT/UPDATE/DELETE 실행
        - 반환: 성공 True / 실패 False
        - 주의: 결과 데이터는 반환하지 않음
        """
        conn = self.connect()
        if conn is None:
            return False

        try:
            cursor = conn.cursor()
            cursor.execute(query, params)
            conn.commit()
            return True
        except Exception as e:
            print(f"[쿼리 실행 오류] {e}")
            return False
        finally:
            # close는 실패하더라도 프로그램이 중단되지 않도록 보호
            try:
                if conn:
                    conn.close()
            except Exception:
                pass

    def fetch_data(self, query, params=None):
        """
        SELECT 실행
        - 반환: list[tuple]
        - 실패 시: []
        """
        conn = self.connect()
        if conn is None:
            return []

        try:
            cursor = conn.cursor()
            cursor.execute(query, params)
            return cursor.fetchall()
        except Exception as e:
            print(f"[데이터 조회 오류] {e}")
            return []
        finally:
            try:
                if conn:
                    conn.close()
            except Exception:
                pass

3. 실행 테스트

1. 콘솔 기반 테스트
   • connect() 성공 여부 확인
   • fetch_data로 간단한 SELECT 테스트
     
     
2. GUI 테스트
   • 버튼 클릭으로 DB 동작 시각화
   • 테이블 출력 확인

4. 학습 주요 포인트

• UI 코드에서 SQL을 구현하지 않고, DatabaseManager 클래스로 분리한다.
• execute_query()는 "변경 쿼리 + commit + 성공/실패 반환" 패턴을 고정한다.
• fetch_data()는 "조회 쿼리 + 결과 리스트 반환" 패턴을 고정한다.
• 연결은 열었으면 반드시 닫는다(finally: conn.close()).
• 파라미터 바인딩(%s, params)을 기본 습관으로 사용한다.

단계별 완성 파일

5. 구현 - 데이터 접근 영역(DB).zip

0.00MB