포스트

[Python 100일 챌린지] Day 78 - Python과 SQLite 연동 심화

게시
Day 78 - Python과 SQLite 연동 심화
Day 78 - Python과 SQLite 연동 심화
By YonYonWare
27 분읽는 시간
[Python 100일 챌린지] Day 78 - Python과 SQLite 연동 심화

Python에서 안전하고 효율적으로 데이터베이스를 다루는 방법을 배워봐요! 🔒

sqlite3 모듈의 고급 기능을 활용하면 SQL Injection을 방지하고, 트랜잭션을 관리하고, 대량 데이터를 효율적으로 처리할 수 있어요. 실무에서 반드시 알아야 할 안전한 DB 연동 패턴들을 익혀봐요!

(35분 완독 ⭐⭐⭐)

🎯 오늘의 학습 목표

📚 사전 지식

🎯 학습 목표 1: 파라미터화된 쿼리로 SQL Injection 방지하기

한 줄 설명

파라미터화된 쿼리 = SQL과 데이터를 분리하여 보안 취약점 차단

사용자 입력을 직접 SQL에 넣지 않고 ? 플레이스홀더를 사용하면 SQL Injection 공격을 막을 수 있어요!

❌ 위험한 방법: 문자열 포맷팅

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import sqlite3

conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# 테이블 생성
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY,
    username TEXT UNIQUE,
    email TEXT,
    age INTEGER
)
''')

# ❌ 절대 이렇게 하지 마세요! (SQL Injection 취약)
username = input("사용자명 입력: ")
# 만약 username = "admin' OR '1'='1" 입력하면?
query = f"SELECT * FROM users WHERE username = '{username}'"
cursor.execute(query)  # 모든 사용자 정보 노출!

왜 위험한가요? 악의적인 사용자가 admin' OR '1'='1을 입력하면:

1
2

SELECT * FROM users WHERE username = 'admin' OR '1'='1'
-- '1'='1'은 항상 참이므로 모든 사용자 정보가 조회됨!

✅ 안전한 방법: 파라미터화된 쿼리

1
2
3
4
5
6
7
8
9
10
11

# ✅ 올바른 방법: ? 플레이스홀더 사용
username = input("사용자명 입력: ")

# ? 자리에 안전하게 값이 대입됨
cursor.execute(
    'SELECT * FROM users WHERE username = ?',
    (username,)  # 튜플로 전달 (쉼표 주의!)
)

# 악의적인 입력도 단순 문자열로 처리됨
# 'admin\' OR \'1\'=\'1' 문자 그대로 검색

파라미터 전달 방법 3가지

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 1. 위치 기반 (? 사용) - 가장 일반적
cursor.execute(
    'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
    ('hong', '[email protected]', 25)
)

# 2. 이름 기반 (:name 사용) - 가독성 좋음
cursor.execute(
    'INSERT INTO users (username, email, age) VALUES (:name, :email, :age)',
    {'name': 'kim', 'email': '[email protected]', 'age': 30}
)

# 3. 여러 행 동시 삽입 (executemany)
users_data = [
    ('user1', '[email protected]', 20),
    ('user2', '[email protected]', 25),
    ('user3', '[email protected]', 30),
]
cursor.executemany(
    'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
    users_data
)
conn.commit()

실전 예제: 안전한 로그인 함수

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

def login(username, password):
    """안전한 로그인 검증"""
    cursor.execute(
        'SELECT * FROM users WHERE username = ? AND password = ?',
        (username, password)
    )
    user = cursor.fetchone()

    if user:
        print(f"로그인 성공! 환영합니다, {user[1]}님!")
        return True
    else:
        print("로그인 실패! 사용자명 또는 비밀번호가 틀렸습니다.")
        return False

# 사용
login('hong', '1234')

실전 예제: 안전한 검색 함수

1
2
3
4
5
6
7
8
9
10
11
12
13
14

def search_users(keyword):
    """사용자명으로 검색 (LIKE 사용)"""
    cursor.execute(
        'SELECT username, email FROM users WHERE username LIKE ?',
        (f'%{keyword}%',)  # %는 Python에서 처리
    )

    results = cursor.fetchall()
    print(f"\n검색 결과 ({len(results)}건):")
    for user in results:
        print(f"  - {user[0]}: {user[1]}")

# 사용
search_users('user')

🎯 학습 목표 2: fetch 메서드로 결과 가져오기

한 줄 설명

fetch 메서드 = 쿼리 결과를 가져오는 3가지 방법

필요에 따라 한 행씩, 여러 행씩, 또는 전체를 가져올 수 있어요!

1. fetchone() - 한 행만 가져오기

1
2
3
4
5
6
7
8
9
10
11

# 가장 어린 사용자 한 명
cursor.execute('SELECT username, age FROM users ORDER BY age LIMIT 1')
user = cursor.fetchone()

if user:
    print(f"가장 어린 사용자: {user[0]} ({user[1]}세)")
else:
    print("사용자가 없습니다.")

# 반환값: 튜플 하나 또는 None
# 예: ('user1', 20) 또는 None

2. fetchmany(size) - 여러 행 가져오기

1
2
3
4
5
6
7
8
9
10
11
12
13

# 상위 3명씩 가져오기
cursor.execute('SELECT username, age FROM users ORDER BY age')

while True:
    rows = cursor.fetchmany(3)  # 3개씩
    if not rows:
        break

    print(f"\n=== 다음 {len(rows)}명 ===")
    for user in rows:
        print(f"{user[0]}: {user[1]}세")

# 메모리 효율적인 대량 데이터 처리에 유용

3. fetchall() - 전체 결과 가져오기

1
2
3
4
5
6
7
8
9
10

# 모든 사용자 조회
cursor.execute('SELECT username, email, age FROM users')
all_users = cursor.fetchall()

print(f"\n=== 전체 사용자 ({len(all_users)}명) ===")
for user in all_users:
    print(f"{user[0]}: {user[1]}, {user[2]}세")

# 반환값: 튜플의 리스트
# 예: [('hong', '[email protected]', 25), ...]

Row Factory로 딕셔너리처럼 사용하기

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# Row Factory 설정 (컬럼명으로 접근 가능)
conn.row_factory = sqlite3.Row

cursor.execute('SELECT * FROM users')
users = cursor.fetchall()

print("\n=== Row Factory 사용 ===")
for user in users:
    # 인덱스로도 접근 가능
    print(f"ID: {user[0]}")

    # 컬럼명으로도 접근 가능! (더 명확함)
    print(f"사용자: {user['username']}")
    print(f"이메일: {user['email']}")
    print(f"나이: {user['age']}\n")

실전 예제: 페이징 처리

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

def get_users_page(page=1, per_page=10):
    """페이징 처리된 사용자 목록"""
    offset = (page - 1) * per_page

    # 전체 개수
    cursor.execute('SELECT COUNT(*) FROM users')
    total = cursor.fetchone()[0]

    # 페이지 데이터
    cursor.execute(
        'SELECT username, email FROM users LIMIT ? OFFSET ?',
        (per_page, offset)
    )
    users = cursor.fetchall()

    total_pages = (total + per_page - 1) // per_page

    print(f"\n=== 페이지 {page}/{total_pages} (전체 {total}명) ===")
    for i, user in enumerate(users, start=offset + 1):
        print(f"{i}. {user[0]}: {user[1]}")

    return users

# 사용
get_users_page(page=1, per_page=5)
get_users_page(page=2, per_page=5)

🎯 학습 목표 3: 대량 데이터 처리와 executemany()

한 줄 설명

executemany() = 여러 행을 한 번에 삽입/수정하는 효율적인 방법

반복문으로 execute() 호출하는 것보다 훨씬 빠르고 효율적이에요!

executemany()로 대량 삽입

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# ❌ 비효율적: 반복문으로 하나씩 삽입
users = [
    ('user1', '[email protected]', 20),
    ('user2', '[email protected]', 25),
    # ... 1000개
]

# 느림!
for user in users:
    cursor.execute(
        'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
        user
    )

# ✅ 효율적: executemany() 사용
cursor.executemany(
    'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
    users  # 리스트 전체 전달
)
conn.commit()  # 한 번만 commit

성능 비교 예제

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import time

# 테스트용 대량 데이터 생성
test_data = [(f'user{i}', f'user{i}@example.com', 20 + i % 50) for i in range(1000)]

# 방법 1: 반복문
start = time.time()
for user in test_data:
    cursor.execute(
        'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
        user
    )
conn.commit()
time1 = time.time() - start

# 방법 2: executemany()
start = time.time()
cursor.executemany(
    'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
    test_data
)
conn.commit()
time2 = time.time() - start

print(f"반복문: {time1:.3f}초")
print(f"executemany(): {time2:.3f}초")
print(f"성능 향상: {time1/time2:.1f}배 빠름!")

# 출력 예:
# 반복문: 0.523초
# executemany(): 0.045초
# 성능 향상: 11.6배 빠름!

실전 예제: CSV 파일 임포트

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

import csv

def import_csv_to_db(csv_file):
    """CSV 파일을 데이터베이스에 임포트"""
    with open(csv_file, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        next(reader)  # 헤더 건너뛰기

        # CSV 데이터를 리스트로 읽기
        data = list(reader)

    # 대량 삽입
    cursor.executemany(
        'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
        data
    )
    conn.commit()

    print(f"✅ {len(data)}개 레코드 임포트 완료!")

# 사용
# import_csv_to_db('users.csv')

executemany()로 대량 업데이트

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 여러 사용자의 나이를 한 번에 업데이트
updates = [
    (26, 'hong'),    # 나이 26세로, username='hong'
    (31, 'kim'),     # 나이 31세로, username='kim'
    (21, 'user1'),   # 나이 21세로, username='user1'
]

cursor.executemany(
    'UPDATE users SET age = ? WHERE username = ?',
    updates
)
conn.commit()

print(f"✅ {cursor.rowcount}개 레코드 업데이트 완료!")

🎯 학습 목표 4: 트랜잭션과 예외 처리 마스터하기

한 줄 설명

트랜잭션 = 여러 SQL 작업을 하나의 단위로 묶어 전부 성공 또는 전부 실패

commit()으로 확정하고, rollback()으로 취소할 수 있어요!

기본 트랜잭션 패턴

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

try:
    # 1. 여러 작업 시작
    cursor.execute('INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
                   ('new_user', '[email protected]', 25))

    cursor.execute('UPDATE users SET age = age + 1 WHERE username = ?',
                   ('hong',))

    # 2. 모든 작업이 성공하면 commit
    conn.commit()
    print("✅ 트랜잭션 성공!")

except sqlite3.Error as e:
    # 3. 오류 발생 시 rollback (모든 작업 취소)
    conn.rollback()
    print(f"❌ 트랜잭션 실패: {e}")

context manager (with 문) 활용

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# ✅ 추천: with 문으로 자동 commit/rollback
def transfer_money(from_user, to_user, amount):
    """계좌 이체 시뮬레이션 (트랜잭션)"""
    try:
        with conn:  # 자동으로 commit/rollback 처리
            # from_user 잔액 감소
            cursor.execute(
                'UPDATE accounts SET balance = balance - ? WHERE username = ?',
                (amount, from_user)
            )

            # to_user 잔액 증가
            cursor.execute(
                'UPDATE accounts SET balance = balance + ? WHERE username = ?',
                (amount, to_user)
            )

            print(f"✅ {from_user} → {to_user}: {amount}원 이체 성공!")

    except sqlite3.Error as e:
        print(f"❌ 이체 실패: {e}")

# 사용
transfer_money('hong', 'kim', 10000)

명시적 트랜잭션 제어

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 자동 commit 끄기
conn.isolation_level = None  # autocommit 모드

# 수동 트랜잭션
cursor.execute('BEGIN')  # 트랜잭션 시작

try:
    cursor.execute('INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
                   ('test_user', '[email protected]', 20))

    cursor.execute('DELETE FROM users WHERE age < 18')

    cursor.execute('COMMIT')  # 확정
    print("✅ 트랜잭션 커밋 완료!")

except sqlite3.Error as e:
    cursor.execute('ROLLBACK')  # 취소
    print(f"❌ 트랜잭션 롤백: {e}")

예외 처리 모범 사례

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

def safe_insert_user(username, email, age):
    """안전한 사용자 추가 (예외 처리 포함)"""
    try:
        cursor.execute(
            'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
            (username, email, age)
        )
        conn.commit()
        print(f"✅ 사용자 '{username}' 추가 완료!")
        return True

    except sqlite3.IntegrityError:
        # UNIQUE 제약조건 위반 (중복 사용자명)
        conn.rollback()
        print(f"❌ 오류: '{username}'은 이미 존재합니다.")
        return False

    except sqlite3.OperationalError as e:
        # 테이블 없음, 컬럼 오류 등
        conn.rollback()
        print(f"❌ 데이터베이스 오류: {e}")
        return False

    except Exception as e:
        # 기타 예외
        conn.rollback()
        print(f"❌ 예상치 못한 오류: {e}")
        return False

# 사용
safe_insert_user('hong', '[email protected]', 25)  # 성공
safe_insert_user('hong', '[email protected]', 30)  # 중복 오류

실전 예제: 데이터베이스 헬퍼 클래스

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

class DatabaseHelper:
    """안전한 데이터베이스 작업 헬퍼"""

    def __init__(self, db_name):
        self.conn = sqlite3.connect(db_name)
        self.conn.row_factory = sqlite3.Row
        self.cursor = self.conn.cursor()

    def __enter__(self):
        """with 문 시작"""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """with 문 종료 (자동 정리)"""
        if exc_type is None:
            self.conn.commit()  # 오류 없으면 commit
        else:
            self.conn.rollback()  # 오류 있으면 rollback
        self.conn.close()

    def execute(self, query, params=()):
        """안전한 쿼리 실행"""
        try:
            self.cursor.execute(query, params)
            return self.cursor
        except sqlite3.Error as e:
            print(f"❌ 쿼리 오류: {e}")
            raise

    def executemany(self, query, params_list):
        """대량 작업 실행"""
        try:
            self.cursor.executemany(query, params_list)
            return self.cursor
        except sqlite3.Error as e:
            print(f"❌ 대량 작업 오류: {e}")
            raise

    def fetchone(self, query, params=()):
        """한 행 조회"""
        self.execute(query, params)
        return self.cursor.fetchone()

    def fetchall(self, query, params=()):
        """전체 조회"""
        self.execute(query, params)
        return self.cursor.fetchall()

# 사용 예제
with DatabaseHelper('mydb.db') as db:
    # 사용자 추가
    db.execute(
        'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
        ('test', '[email protected]', 25)
    )

    # 사용자 조회
    user = db.fetchone(
        'SELECT * FROM users WHERE username = ?',
        ('test',)
    )
    print(f"조회: {user['username']}, {user['email']}")

# with 블록 종료 시 자동으로 commit/close

💡 실전 팁 & 주의사항

💡 Tip 1: 항상 파라미터화된 쿼리 사용

1
2
3
4
5
6

# ❌ 절대 금지
username = input("사용자명: ")
query = f"SELECT * FROM users WHERE username = '{username}'"

# ✅ 항상 이렇게
cursor.execute('SELECT * FROM users WHERE username = ?', (username,))

💡 Tip 2: commit() 잊지 않기

1
2
3
4
5
6
7

# ❌ commit 없으면 데이터 저장 안 됨!
cursor.execute('INSERT INTO users VALUES (?, ?, ?)', ('hong', '[email protected]', 25))
# 저장 안 됨!

# ✅ commit 필수!
cursor.execute('INSERT INTO users VALUES (?, ?, ?)', ('hong', '[email protected]', 25))
conn.commit()  # 이제 저장됨!

💡 Tip 3: 연결 종료는 마지막에

1
2
3
4
5
6
7
8
9
10
11
12

# ✅ 작업 완료 후 닫기
try:
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()
finally:
    conn.close()  # 항상 닫기

# 또는 with 문 사용 (자동 닫기)
with sqlite3.connect('mydb.db') as conn:
    cursor = conn.cursor()
    # 작업...
# 자동으로 닫힘

💡 Tip 4: 단일 값 튜플 주의

1
2
3
4
5
6
7
8

# ❌ 오류: 괄호만 있으면 튜플이 아님
cursor.execute('SELECT * FROM users WHERE username = ?', ('hong'))

# ✅ 쉼표 필수!
cursor.execute('SELECT * FROM users WHERE username = ?', ('hong',))

# 또는 리스트도 가능
cursor.execute('SELECT * FROM users WHERE username = ?', ['hong'])

🧪 연습 문제

문제 1: 안전한 게시판 시스템

과제: SQL Injection에 안전한 간단한 게시판 CRUD를 구현하세요.

요구사항:

  1. posts 테이블 생성 (id, title, content, author, created_at)
  2. 게시글 작성 함수 (파라미터화된 쿼리)
  3. 게시글 검색 함수 (제목 LIKE 검색)
  4. 게시글 삭제 함수 (트랜잭션)
  5. 전체 게시글 조회 (fetchall)
💡 힌트
  • datetime.now()로 현재 시간 저장
  • LIKE 검색: WHERE title LIKE ?, (f'%{keyword}%',)
  • DELETE 시 try-except로 예외 처리
  • fetchall()로 전체 조회 후 for 반복
✅ 정답 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

import sqlite3
from datetime import datetime

class BoardSystem:
    def __init__(self, db_name='board.db'):
        self.conn = sqlite3.connect(db_name)
        self.conn.row_factory = sqlite3.Row
        self.cursor = self.conn.cursor()
        self._create_table()

    def _create_table(self):
        """테이블 생성"""
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS posts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                content TEXT,
                author TEXT NOT NULL,
                created_at TEXT NOT NULL
            )
        ''')
        self.conn.commit()

    def create_post(self, title, content, author):
        """게시글 작성"""
        try:
            self.cursor.execute('''
                INSERT INTO posts (title, content, author, created_at)
                VALUES (?, ?, ?, ?)
            ''', (title, content, author, datetime.now().isoformat()))
            self.conn.commit()
            print(f"✅ 게시글 '{title}' 작성 완료!")
            return self.cursor.lastrowid
        except sqlite3.Error as e:
            self.conn.rollback()
            print(f"❌ 작성 실패: {e}")
            return None

    def search_posts(self, keyword):
        """게시글 검색"""
        self.cursor.execute('''
            SELECT * FROM posts
            WHERE title LIKE ? OR content LIKE ?
            ORDER BY created_at DESC
        ''', (f'%{keyword}%', f'%{keyword}%'))

        posts = self.cursor.fetchall()
        print(f"\n검색 결과: {len(posts)}건")
        for post in posts:
            print(f"[{post['id']}] {post['title']} - {post['author']}")

        return posts

    def delete_post(self, post_id):
        """게시글 삭제"""
        try:
            self.cursor.execute('DELETE FROM posts WHERE id = ?', (post_id,))
            if self.cursor.rowcount > 0:
                self.conn.commit()
                print(f"✅ 게시글 {post_id} 삭제 완료!")
                return True
            else:
                print(f"❌ 게시글 {post_id}를 찾을 수 없습니다.")
                return False
        except sqlite3.Error as e:
            self.conn.rollback()
            print(f"❌ 삭제 실패: {e}")
            return False

    def list_all_posts(self):
        """전체 게시글 조회"""
        self.cursor.execute('SELECT * FROM posts ORDER BY created_at DESC')
        posts = self.cursor.fetchall()

        print(f"\n=== 전체 게시글 ({len(posts)}건) ===")
        for post in posts:
            print(f"[{post['id']}] {post['title']}")
            print(f"    작성자: {post['author']}")
            print(f"    작성일: {post['created_at'][:10]}\n")

        return posts

    def close(self):
        """연결 종료"""
        self.conn.close()

# 사용 예제
board = BoardSystem()

# 게시글 작성
board.create_post('Python 강좌', 'Python 기초부터 실전까지', 'hong')
board.create_post('SQL 튜토리얼', 'SQLite 활용법', 'kim')
board.create_post('웹 개발', 'Flask로 웹사이트 만들기', 'lee')

# 검색
board.search_posts('Python')

# 전체 조회
board.list_all_posts()

# 삭제
board.delete_post(1)

# 종료
board.close()

출력:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

✅ 게시글 'Python 강좌' 작성 완료!
✅ 게시글 'SQL 튜토리얼' 작성 완료!
✅ 게시글 '웹 개발' 작성 완료!

검색 결과: 1건
[1] Python 강좌 - hong

=== 전체 게시글 (3건) ===
[3] 웹 개발
    작성자: lee
    작성일: 2025-05-17

[2] SQL 튜토리얼
    작성자: kim
    작성일: 2025-05-17

[1] Python 강좌
    작성자: hong
    작성일: 2025-05-17

✅ 게시글 1 삭제 완료!

문제 2: 대량 데이터 임포트 시스템

과제: CSV 파일에서 데이터를 읽어 효율적으로 DB에 저장하세요.

요구사항:

  1. 1000개 이상의 샘플 데이터 생성 (faker 또는 임의 데이터)
  2. executemany()로 대량 삽입 (성능 측정)
  3. 트랜잭션으로 안전하게 처리
  4. 진행률 표시 (100개마다 출력)
  5. 오류 발생 시 rollback
💡 힌트
  • range(1000)으로 샘플 데이터 생성
  • time.time()으로 시작/종료 시간 측정
  • try-except로 예외 처리
  • executemany() 한 번만 호출
✅ 정답 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

import sqlite3
import time
from random import randint, choice

def generate_sample_data(count=1000):
    """샘플 데이터 생성"""
    names = ['김철수', '이영희', '박민수', '최지혜', '정다은']
    cities = ['서울', '부산', '대구', '인천', '광주']

    data = []
    for i in range(count):
        username = f"user{i+1}"
        email = f"{username}@example.com"
        age = randint(18, 70)
        city = choice(cities)
        data.append((username, email, age, city))

    return data

def import_bulk_data(data, batch_size=100):
    """대량 데이터 임포트"""
    conn = sqlite3.connect('bulk_test.db')
    cursor = conn.cursor()

    # 테이블 생성
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE,
            email TEXT,
            age INTEGER,
            city TEXT
        )
    ''')

    try:
        start_time = time.time()

        # 방법 1: 전체 한 번에 (가장 빠름)
        print(f"📥 {len(data)}개 레코드 임포트 시작...")
        cursor.executemany(
            'INSERT INTO users (username, email, age, city) VALUES (?, ?, ?, ?)',
            data
        )

        conn.commit()
        elapsed = time.time() - start_time

        print(f"✅ 임포트 완료!")
        print(f"   - 레코드 수: {len(data):,}개")
        print(f"   - 소요 시간: {elapsed:.3f}초")
        print(f"   - 초당 처리: {len(data)/elapsed:,.0f}개/초")

        return True

    except sqlite3.IntegrityError as e:
        conn.rollback()
        print(f"❌ 중복 데이터 오류: {e}")
        return False

    except sqlite3.Error as e:
        conn.rollback()
        print(f"❌ 데이터베이스 오류: {e}")
        return False

    finally:
        conn.close()

# 실행
sample_data = generate_sample_data(10000)  # 10000개 생성
import_bulk_data(sample_data)

# 검증
conn = sqlite3.connect('bulk_test.db')
cursor = conn.cursor()

cursor.execute('SELECT COUNT(*) FROM users')
total = cursor.fetchone()[0]
print(f"\n검증: 총 {total:,}개 레코드 저장됨")

cursor.execute('SELECT city, COUNT(*) FROM users GROUP BY city')
print("\n도시별 분포:")
for row in cursor.fetchall():
    print(f"  {row[0]}: {row[1]:,}명")

conn.close()

출력:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

📥 10,000개 레코드 임포트 시작...
✅ 임포트 완료!
   - 레코드 수: 10,000개
   - 소요 시간: 0.127초
   - 초당 처리: 78,740개/초

검증: 총 10,000개 레코드 저장됨

도시별 분포:
  광주: 1,989명
  대구: 2,015명
  부산: 2,012명
  서울: 2,009명
  인천: 1,975명

📝 오늘 배운 내용 정리

  1. 파라미터화된 쿼리: ? 플레이스홀더로 SQL Injection 방지
  2. fetch 메서드: fetchone(), fetchmany(), fetchall()
  3. executemany(): 대량 데이터 효율적 처리
  4. 트랜잭션: commit(), rollback()로 데이터 무결성 보장
  5. 예외 처리: try-except로 안전한 DB 작업
  6. context manager: with 문으로 자동 리소스 관리

🔗 관련 자료

📚 이전 학습

Day 77: SQL 고급 - 집계함수와 GROUP BY ⭐⭐⭐

어제는 COUNT, SUM, AVG 같은 집계함수와 GROUP BY로 데이터를 분석하는 방법을 배웠어요!

📚 다음 학습

Day 79: Python DB 고급 - ORM 개념 ⭐⭐⭐

내일은 SQL 없이 Python 코드만으로 데이터베이스를 다루는 ORM(Object-Relational Mapping)을 배워요!

“안전한 코드는 신뢰할 수 있는 서비스의 기초입니다!” 🔒

Day 78/100 Phase 8: 데이터베이스와 SQL #100DaysOfPython
Python100일챌린지, Python100일챌린지-Phase08
python 100일챌린지 sqlite database crud 100DaysOfPython
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.