📦 fetch 메서드 비교

SELECT를 실행했다고 해서 데이터가 바로 Python으로 넘어오지는 않는다.
커서는 결과를 들고 있고, fetch가 데이터를 꺼낸다.

fetch 방식은 총 세 가지다. 상황에 맞게 선택하지 않으면 메모리 폭탄을 맞는다.

┌────────────┬──────────────────────────────┬─────────────────────┐
│ 메서드       │ 의미                          │ 반환 형태             │
├────────────┼──────────────────────────────┼─────────────────────┤
│ fetchone() │ 1행만 가져오기                  │ 튜플 1개 / None          │
│ fetchall() │ 전부 가져오기                   │ 튜플 리스트               │
│ fetchmany  │ n개씩 나눠서 가져오기             │ 튜플 리스트               │
└────────────┴──────────────────────────────┴─────────────────────┘
  

🥄 fetchone()

결과가 1건임이 보장될 때 사용하는 메서드다.
주로 PK(id) 조회에 사용한다.

cursor.execute("SELECT * FROM users WHERE id = 1")
row = cursor.fetchone()

if row:
    print(row)
else:
    print("결과 없음")

(1, '철수', 'chulsoo@example.com', 25)

결과가 없으면 None을 반환한다.
예외가 아니라 None이라는 점이 중요하다.

🍱 fetchall()

조건에 맞는 모든 행을 한 번에 메모리로 가져온다.

cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()

print(f"총 {len(rows)}건")

for row in rows:
    print(row)

⚠️ 데이터가 많을수록 위험하다

[ DB 결과 100만 건 ]
        ↓
fetchall()
        ↓
Python 메모리 100만 건 적재 💥
  

관리자 페이지, 테스트 데이터처럼 행 수가 적다는 게 확실할 때만 사용한다.

📦 fetchmany(n)

대용량 데이터를 다룰 때 사용하는 메서드다.
결과를 조각 단위로 나눠서 처리한다.

cursor.execute("SELECT * FROM users")

while True:
    rows = cursor.fetchmany(2)
    if not rows:
        break

    for row in rows:
        print(row)

[1] 철수
[2] 영희
---
[3] 훈이
[4] 민수
---
  

로그 처리, 배치 작업, 통계 집계에서 필수적으로 사용됨.

🧾 튜플 결과의 문제점

기본 커서는 결과를 튜플(tuple)로 반환한다.

row = (1, '철수', 'chulsoo@example.com', 25)

print(row[1])  # 철수

이 코드의 문제는 뭘까?

row[1]  ← 1이 뭘 의미하는지 코드만 보고 알 수 없음
  

컬럼 순서를 외워야 하고,
컬럼이 하나라도 바뀌면 전부 수정해야 한다.

🗂 DictCursor — 사람이 읽을 수 있는 결과

DictCursor를 사용하면 결과를 딕셔너리로 받는다.

from pymysql.cursors import DictCursor

conn = pymysql.connect(
    host="서버주소",
    user="사용자명",
    password="비밀번호",
    db="데이터베이스명",
    charset="utf8mb4",
    cursorclass=DictCursor
)
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")

for row in cursor.fetchall():
    print(f"{row['name']} ({row['email']})")

출력 예시:

철수 (chulsoo@example.com)
영희 (younghee@example.com)
훈이 (hooni@example.com)

┌──────────────┬───────────────┐
│ 방식          │ 접근 방법       │
├──────────────┼───────────────┤
│ 튜플          │ row[1]        │
│ DictCursor   │ row['name']   │
└──────────────┴───────────────┘
  

👉 실무에서는 DictCursor가 사실상 표준이다

🛡 SQL Injection 방지

SQL을 문자열로 조합하는 순간, 보안이 무너진다.

# ❌ 절대 쓰지 말 것
name = "철수"
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")

입력값:
'; DROP TABLE users; --
  

실행 SQL:
SELECT * FROM users WHERE name = '';
DROP TABLE users;
  

✅ 파라미터 바인딩

name = "철수"

cursor.execute(
    "SELECT * FROM users WHERE name = %s",
    (name,)
)

pymysql이 자동으로 이스케이프 처리한다.SQL 구조와 값이 분리된다.

┌─────────────┬─────────────────────────┐
│ 방식         │ 안전성                    │
├─────────────┼─────────────────────────┤
│ 문자열 조합    │ ❌ Injection 위험        │
│ %s 바인딩     │ ✅ 안전                  │
└─────────────┴─────────────────────────┘
  

🧱 함수로 정리하는 이유

실무에서 DB 코드를 그대로 쓰는 일은 없다.
반드시 함수로 감싼다.

def get_user_by_id(user_id):
    conn = get_connection()
    cursor = conn.cursor()

    cursor.execute(
        "SELECT * FROM users WHERE id = %s",
        (user_id,)
    )

    user = cursor.fetchone()

    cursor.close()
    conn.close()

    return user

• 재사용 가능
• 테스트 쉬움
• 트랜잭션 관리 쉬움
• 코드 읽기 쉬움

[ 이번 실습에서 훈련하는 것 ]

- SELECT 조건 검색
- fetch 메서드 선택 기준
- 결과가 없을 때의 처리 방식
- DictCursor + %s 바인딩 습관화
  

import pymysql
from pymysql.cursors import DictCursor

def get_connection():
    return pymysql.connect(
        host="서버주소",
        user="사용자명",
        password="비밀번호",
        db="데이터베이스명",
        charset="utf8mb4",
        cursorclass=DictCursor
    )

┌───────────────────────────┐
│ get_connection() 역할      │
├───────────────────────────┤
│ - DB 연결 생성              │
│ - DictCursor 강제          │
│ - 연결 설정 중앙 관리         │
└───────────────────────────┘
  

def get_users_by_age_range(min_age, max_age):
    conn = get_connection()
    cursor = conn.cursor()

    cursor.execute(
        "SELECT * FROM users WHERE age BETWEEN %s AND %s",
        (min_age, max_age)
    )

    users = cursor.fetchall()

    cursor.close()
    conn.close()

    return users

[ 사고 흐름 ]

- 결과가 0건일 수도 있음
- fetchall()은 결과 없으면 []
- 호출하는 쪽에서 len(users)로 판단 가능
  

users = get_users_by_age_range(20, 30)

if not users:
    print("해당 나이대 사용자 없음")
else:
    for user in users:
        print(user["name"], user["age"])

def get_user_by_email(email):
    conn = get_connection()
    cursor = conn.cursor()

    cursor.execute(
        "SELECT * FROM users WHERE email = %s",
        (email,)
    )

    user = cursor.fetchone()

    cursor.close()
    conn.close()

    return user

[ 포인트 ]

- 결과 없음 → None
- 예외 X, None 반환
- 호출부에서 분기 처리
  

user = get_user_by_email("chulsoo@example.com")

if user is None:
    print("해당 이메일 사용자 없음")
else:
    print(user["name"], user["email"])

def count_users():
    conn = get_connection()
    cursor = conn.cursor()

    cursor.execute("SELECT COUNT(*) AS cnt FROM users")
    result = cursor.fetchone()

    cursor.close()
    conn.close()

    return result["cnt"]

┌───────────────────────────┐
│ COUNT 쿼리 특징             │
├───────────────────────────┤
│ - 결과는 무조건 1행           │
│ - 값이 0일 수는 있음          │
│ - None 아님                │
└───────────────────────────┘
  

total = count_users()
print(f"전체 사용자 수: {total}")

┌───────────────────────────┬────────────────────┐
│ 상황                       │ 반환 값              │
├───────────────────────────┼────────────────────┤
│ fetchone 결과 없음          │ None               │
│ fetchall 결과 없음          │ [] (빈 리스트)       │
│ COUNT(*)                  │ 0                  │
└───────────────────────────┴────────────────────┘
  

✅ 핵심 요약

✔ fetchone → 단건 조회
✔ fetchall → 소량 데이터
✔ fetchmany → 대용량 데이터
✔ DictCursor → 실무 필수
✔ %s 바인딩 → 보안 필수