[Python 100일 챌린지] Day 79 - Python DB 고급: ORM 개념

SQL은 복잡하고 번거롭다고요? 🤔

오늘은 데이터베이스를 마치 파이썬 객체처럼 다룰 수 있는 마법 같은 기술, ORM을 배워봐요! SQL 쿼리를 일일이 작성하지 않아도 파이썬 코드만으로 데이터베이스를 자유자재로 조작할 수 있어요! SQLAlchemy를 사용해서 실전 예제로 완벽하게 마스터해봐요! 🚀

(40분 완독 ⭐⭐⭐⭐⭐)

🎯 오늘의 학습 목표

Phase 8의 마지막 개념 학습! ORM(Object-Relational Mapping)을 완벽하게 이해해요!

• ✅ ORM이 무엇인지, 왜 필요한지 이해하기
• ✅ SQL vs ORM 비교로 장단점 파악하기
• ✅ SQLAlchemy로 실전 ORM 구현하기
• ✅ 언제 ORM을 쓰고, 언제 Raw SQL을 써야 하는지 판단하기

📚 사전 지식

필수 선행 학습:

• Day 71: SQLite 기초 - 데이터베이스 개념
• Day 78: Python과 SQLite 연동 - sqlite3 모듈 사용법

도움이 되는 개념:

• Day 69: 타입 힌팅과 데이터클래스 - 클래스 설계
• Day 55: 클래스 심화 - OOP 개념

---

🎯 학습 목표 1: ORM이란? (Object-Relational Mapping)

ORM의 핵심 개념 💡

ORM은 객체 지향 프로그래밍 언어와 관계형 데이터베이스를 자동으로 연결해주는 기술이에요!

파이썬 클래스  ⟷  데이터베이스 테이블
객체 인스턴스  ⟷  테이블의 행(Row)
클래스 속성    ⟷  테이블의 컬럼(Column)

왜 ORM이 필요할까요? 🤔

문제 1: SQL과 파이썬 코드의 불일치

# Raw SQL - 문자열로 작성
query = "SELECT * FROM users WHERE age > 20"

# 결과는 튜플이나 딕셔너리
# 타입 힌팅도 안 되고, IDE 자동완성도 안 됨!

문제 2: 반복적인 보일러플레이트 코드

# 매번 이런 코드를 작성해야 함
cursor.execute("INSERT INTO users VALUES (?, ?, ?)", (id, name, age))
conn.commit()

문제 3: 데이터베이스 변경 시 코드 수정 어려움

# 컬럼명 변경 시 모든 쿼리를 찾아서 수정해야 함!
# 오타가 있어도 실행 전까지 알 수 없음!

ORM으로 해결! ✨

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

# 1. 파이썬 클래스로 테이블 정의
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

# 2. 파이썬 객체처럼 사용!
user = User(name='김철수', age=25)
session.add(user)
session.commit()

# 3. 쿼리도 파이썬 코드로!
users = session.query(User).filter(User.age > 20).all()

ORM의 장점:

• ✅ SQL을 몰라도 데이터베이스 조작 가능
• ✅ IDE 자동완성, 타입 체크 지원
• ✅ 오타 방지, 리팩토링 쉬움
• ✅ 데이터베이스 변경 시 코드 수정 최소화
• ✅ 재사용 가능한 비즈니스 로직

---

🎯 학습 목표 2: SQL vs ORM 비교

실전 예제로 비교해봐요! 🔍

예제: 사용자 관리 시스템

Raw SQL 방식 📝

import sqlite3

# 연결
conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# 1. 테이블 생성
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# 2. 데이터 삽입
cursor.execute(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    ('김철수', '[email protected]', 25)
)
conn.commit()

# 3. 조회
cursor.execute("SELECT * FROM users WHERE age > ?", (20,))
rows = cursor.fetchall()

for row in rows:
    # 튜플이라 인덱스로 접근해야 함!
    print(f"이름: {row[1]}, 이메일: {row[2]}")

# 4. 수정
cursor.execute(
    "UPDATE users SET age = ? WHERE name = ?",
    (26, '김철수')
)
conn.commit()

# 5. 삭제
cursor.execute("DELETE FROM users WHERE age < ?", (18,))
conn.commit()

cursor.close()
conn.close()

단점:

• ❌ SQL 문자열로 작성 (오타 위험!)
• ❌ 결과가 튜플이라 다루기 불편
• ❌ 타입 체크 안 됨
• ❌ 반복적인 commit/close 코드

ORM (SQLAlchemy) 방식 🚀

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

Base = declarative_base()

# 1. 모델 정의 (테이블 구조)
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.now)

    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}')>"

# 엔진 및 세션 설정
engine = create_engine('sqlite:///users.db')
Base.metadata.create_all(engine)  # 테이블 자동 생성!
Session = sessionmaker(bind=engine)
session = Session()

# 2. 데이터 삽입 (파이썬 객체처럼!)
user = User(name='김철수', email='[email protected]', age=25)
session.add(user)
session.commit()

# 3. 조회 (파이썬 코드로!)
users = session.query(User).filter(User.age > 20).all()

for user in users:
    # 객체 속성으로 접근!
    print(f"이름: {user.name}, 이메일: {user.email}")

# 4. 수정 (파이썬 객체 수정하듯이!)
user = session.query(User).filter(User.name == '김철수').first()
user.age = 26
session.commit()

# 5. 삭제
session.query(User).filter(User.age < 18).delete()
session.commit()

session.close()

장점:

• ✅ 파이썬 클래스로 직관적 표현
• ✅ IDE 자동완성 지원
• ✅ 타입 안전성
• ✅ 가독성 높음

비교표 📊

항목	Raw SQL	ORM
학습 곡선	SQL 필수	파이썬만 알면 OK
코드 가독성	낮음	높음
유지보수성	어려움	쉬움
성능	최적화 가능	약간 느릴 수 있음
복잡한 쿼리	유연함	제한적일 수 있음
타입 안전성	없음	있음

---

🎯 학습 목표 3: SQLAlchemy 소개 및 설치

SQLAlchemy란? 🛠️

파이썬에서 가장 인기 있는 ORM 라이브러리예요!

특징:

• 🎯 다양한 데이터베이스 지원: SQLite, MySQL, PostgreSQL, Oracle 등
• 🔄 유연성: ORM과 Raw SQL 모두 사용 가능
• 📦 풍부한 기능: 관계 설정, 트랜잭션, 마이그레이션 등
• 🚀 성능 최적화: Lazy Loading, Eager Loading 등

설치하기 📥

# 기본 설치
pip install sqlalchemy

# SQLite는 파이썬 내장이므로 추가 설치 불필요!
# MySQL 사용 시
pip install pymysql

# PostgreSQL 사용 시
pip install psycopg2-binary

기본 구조 이해하기 🏗️

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 1. 엔진: 데이터베이스 연결 관리
engine = create_engine('sqlite:///mydb.db', echo=True)
# echo=True: SQL 쿼리 로그 출력 (디버깅용)

# 2. Base: 모든 모델의 부모 클래스
Base = declarative_base()

# 3. 세션: 데이터베이스 작업 수행
Session = sessionmaker(bind=engine)
session = Session()

핵심 컴포넌트:

Engine (엔진)
  ↓
Base (베이스 클래스)
  ↓
Model (모델 클래스들)
  ↓
Session (세션)
  ↓
데이터베이스 작업 (CRUD)

---

🎯 학습 목표 4: 모델 클래스 정의

기본 모델 정의 🏛️

from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'  # 테이블 이름

    # 컬럼 정의
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)  # 필수 필드
    price = Column(Float, default=0.0)
    stock = Column(Integer, default=0)
    is_available = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)

    # 객체를 문자열로 표현
    def __repr__(self):
        return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"

다양한 컬럼 타입 🎨

from sqlalchemy import (
    Integer,    # 정수
    String,     # 문자열
    Float,      # 실수
    Boolean,    # 참/거짓
    DateTime,   # 날짜/시간
    Date,       # 날짜
    Time,       # 시간
    Text,       # 긴 텍스트
    LargeBinary # 바이너리 데이터
)

class Example(Base):
    __tablename__ = 'examples'

    id = Column(Integer, primary_key=True)
    title = Column(String(200))        # 최대 200자
    description = Column(Text)         # 제한 없음
    price = Column(Float)
    is_active = Column(Boolean)
    created_date = Column(Date)
    created_time = Column(Time)
    created_at = Column(DateTime)
    image_data = Column(LargeBinary)   # 이미지 등

컬럼 제약조건 ⚙️

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)

    # nullable: NULL 허용 여부
    username = Column(String(50), nullable=False)

    # unique: 중복 불가
    email = Column(String(100), unique=True, nullable=False)

    # default: 기본값
    age = Column(Integer, default=0)
    status = Column(String(20), default='active')

    # index: 인덱스 생성 (검색 속도 향상)
    phone = Column(String(20), index=True)

실전 예제: 블로그 포스트 모델 📝

from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
from datetime import datetime

class BlogPost(Base):
    __tablename__ = 'blog_posts'

    # 기본 정보
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False, index=True)
    content = Column(Text, nullable=False)

    # 메타 정보
    author = Column(String(100), nullable=False)
    category = Column(String(50), default='일반')
    tags = Column(String(200))  # 쉼표로 구분: "python,orm,db"

    # 상태
    is_published = Column(Boolean, default=False)
    view_count = Column(Integer, default=0)

    # 시간 정보
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

    def __repr__(self):
        return f"<BlogPost(id={self.id}, title='{self.title}', author='{self.author}')>"

    # 편의 메서드
    def publish(self):
        """포스트 발행"""
        self.is_published = True

    def add_view(self):
        """조회수 증가"""
        self.view_count += 1

---

🎯 학습 목표 5: CRUD with ORM

Create (생성) ➕

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 엔진 및 세션 설정
engine = create_engine('sqlite:///blog.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 1. 단일 객체 추가
post = BlogPost(
    title='SQLAlchemy ORM 완벽 가이드',
    content='ORM을 사용하면 데이터베이스 작업이 정말 쉬워져요!',
    author='김철수',
    category='기술'
)
session.add(post)
session.commit()

print(f"생성된 포스트 ID: {post.id}")  # 자동으로 할당된 ID!

# 2. 여러 객체 한 번에 추가
posts = [
    BlogPost(title='Python 기초', content='...', author='이영희'),
    BlogPost(title='Django 시작하기', content='...', author='박민수'),
    BlogPost(title='Flask vs FastAPI', content='...', author='김철수')
]
session.add_all(posts)
session.commit()

Read (조회) 🔍

# 1. 모든 데이터 가져오기
all_posts = session.query(BlogPost).all()
print(f"총 {len(all_posts)}개 포스트")

# 2. 첫 번째 데이터
first_post = session.query(BlogPost).first()
print(first_post.title)

# 3. 조건으로 필터링
# 특정 저자의 포스트
kim_posts = session.query(BlogPost).filter(BlogPost.author == '김철수').all()

# 여러 조건 (AND)
published_posts = session.query(BlogPost).filter(
    BlogPost.is_published == True,
    BlogPost.category == '기술'
).all()

# OR 조건
from sqlalchemy import or_
popular_posts = session.query(BlogPost).filter(
    or_(
        BlogPost.view_count > 100,
        BlogPost.is_published == True
    )
).all()

# 4. 정렬
latest_posts = session.query(BlogPost).order_by(
    BlogPost.created_at.desc()
).limit(5).all()

# 5. 특정 컬럼만 조회
titles = session.query(BlogPost.title, BlogPost.author).all()
for title, author in titles:
    print(f"{title} by {author}")

# 6. 개수 세기
count = session.query(BlogPost).filter(BlogPost.is_published == True).count()
print(f"발행된 포스트: {count}개")

Update (수정) ✏️

# 1. 객체 가져와서 수정
post = session.query(BlogPost).filter(BlogPost.id == 1).first()
if post:
    post.title = '수정된 제목'
    post.content = '수정된 내용'
    session.commit()

# 2. 일괄 수정
session.query(BlogPost).filter(
    BlogPost.author == '김철수'
).update({
    BlogPost.category: '기술/Python'
})
session.commit()

# 3. 편의 메서드 사용
post = session.query(BlogPost).first()
post.publish()  # is_published = True 설정
post.add_view()  # view_count 증가
session.commit()

Delete (삭제) 🗑️

# 1. 객체 가져와서 삭제
post = session.query(BlogPost).filter(BlogPost.id == 1).first()
if post:
    session.delete(post)
    session.commit()

# 2. 조건으로 일괄 삭제
session.query(BlogPost).filter(
    BlogPost.is_published == False,
    BlogPost.view_count == 0
).delete()
session.commit()

실전 예제: 완전한 CRUD 💪

def create_post(title, content, author, category='일반'):
    """포스트 생성"""
    post = BlogPost(
        title=title,
        content=content,
        author=author,
        category=category
    )
    session.add(post)
    session.commit()
    return post

def get_post_by_id(post_id):
    """ID로 포스트 조회"""
    return session.query(BlogPost).filter(BlogPost.id == post_id).first()

def get_posts_by_author(author):
    """저자별 포스트 목록"""
    return session.query(BlogPost).filter(BlogPost.author == author).all()

def update_post(post_id, **kwargs):
    """포스트 수정"""
    post = get_post_by_id(post_id)
    if post:
        for key, value in kwargs.items():
            setattr(post, key, value)
        session.commit()
        return post
    return None

def delete_post(post_id):
    """포스트 삭제"""
    post = get_post_by_id(post_id)
    if post:
        session.delete(post)
        session.commit()
        return True
    return False

# 사용 예시
post = create_post('ORM 마스터하기', 'ORM 너무 쉬워요!', '김철수', '기술')
print(f"생성: {post}")

posts = get_posts_by_author('김철수')
print(f"김철수의 포스트: {len(posts)}개")

update_post(post.id, title='ORM 완벽 가이드')
print("수정 완료!")

delete_post(post.id)
print("삭제 완료!")

---

🎯 학습 목표 6: ORM의 장단점

장점 ✅

1. 생산성 향상 🚀

# SQL: 복잡한 쿼리 작성
query = """
    SELECT users.name, COUNT(posts.id) as post_count
    FROM users
    LEFT JOIN posts ON users.id = posts.user_id
    GROUP BY users.id
    HAVING COUNT(posts.id) > 5
"""

# ORM: 파이썬 코드로 직관적
from sqlalchemy import func
session.query(
    User.name,
    func.count(Post.id).label('post_count')
).join(Post).group_by(User.id).having(
    func.count(Post.id) > 5
).all()

2. 유지보수성 🔧

• 컬럼명 변경 시 IDE가 자동으로 찾아줌
• 리팩토링 도구 사용 가능
• 타입 체크로 버그 사전 방지

3. 데이터베이스 독립성 🌐

# SQLite에서 PostgreSQL로 변경해도 코드는 그대로!
# engine = create_engine('sqlite:///mydb.db')
engine = create_engine('postgresql://user:pass@localhost/mydb')

4. 보안 🛡️

• SQL Injection 자동 방지
• 파라미터 바인딩 자동 처리

단점 ❌

1. 성능 오버헤드 🐌

# N+1 문제: 느려질 수 있음!
users = session.query(User).all()
for user in users:
    # 각 사용자마다 추가 쿼리 발생!
    posts = user.posts  # 비효율적!

# 해결: Eager Loading
users = session.query(User).options(joinedload(User.posts)).all()

2. 복잡한 쿼리 제한 🔒

# 매우 복잡한 쿼리는 Raw SQL이 나을 수 있음
result = session.execute("""
    WITH RECURSIVE ...
    SELECT ... (복잡한 쿼리)
""")

3. 학습 곡선 📈

• ORM 자체를 배워야 함
• 내부 동작 이해 필요

---

🎯 학습 목표 7: 언제 ORM, 언제 Raw SQL?

결정 가이드 🎯

ORM을 사용하세요! ✅

1. CRUD 중심의 애플리케이션

# 사용자 관리, 블로그, 쇼핑몰 등
user = User(name='김철수', email='[email protected]')
session.add(user)
session.commit()

2. 빠른 프로토타이핑

• 스타트업, MVP 개발
• 코드 작성 속도가 중요할 때

3. 팀 협업

• 파이썬 개발자만 있는 팀
• SQL 전문가 없을 때

4. 데이터베이스 마이그레이션 가능성

• SQLite → PostgreSQL 등 변경 예정

Raw SQL을 사용하세요! 📝

1. 복잡한 분석 쿼리

-- 윈도우 함수, 재귀 쿼리 등
WITH RECURSIVE category_tree AS (
    SELECT id, name, parent_id, 1 as level
    FROM categories
    WHERE parent_id IS NULL
    UNION ALL
    SELECT c.id, c.name, c.parent_id, ct.level + 1
    FROM categories c
    INNER JOIN category_tree ct ON c.parent_id = ct.id
)
SELECT * FROM category_tree;

2. 성능이 매우 중요한 경우

• 대용량 데이터 처리
• 실시간 시스템
• 최적화된 쿼리 필요

3. 데이터베이스 특화 기능

-- PostgreSQL의 Full Text Search
SELECT * FROM articles
WHERE to_tsvector('korean', content) @@ to_tsquery('파이썬');

하이브리드 접근! 🎭

가장 현명한 선택: 둘 다 사용하기!

# 기본적으로 ORM 사용
user = session.query(User).filter(User.id == 1).first()

# 복잡한 통계는 Raw SQL
result = session.execute("""
    SELECT
        DATE(created_at) as date,
        COUNT(*) as count,
        AVG(view_count) as avg_views
    FROM blog_posts
    GROUP BY DATE(created_at)
    ORDER BY date DESC
    LIMIT 30
""")

for row in result:
    print(f"{row.date}: {row.count}개 포스트, 평균 조회수 {row.avg_views}")

---

💡 실전 팁 & 주의사항

🎯 ORM 사용 팁

1. 세션 관리

# 좋은 예: with 문 사용
from contextlib import contextmanager

@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

# 사용
with get_session() as session:
    user = User(name='김철수')
    session.add(user)

2. N+1 문제 피하기

# 나쁜 예
users = session.query(User).all()
for user in users:
    print(user.posts)  # 각 유저마다 쿼리 발생!

# 좋은 예: Eager Loading
from sqlalchemy.orm import joinedload
users = session.query(User).options(joinedload(User.posts)).all()

3. 대량 삽입 최적화

# 느림
for data in large_data_list:
    session.add(BlogPost(**data))
    session.commit()  # 매번 커밋!

# 빠름
session.bulk_insert_mappings(BlogPost, large_data_list)
session.commit()

⚠️ 주의사항

1. 트랜잭션 관리

try:
    session.add(user)
    session.add(post)
    session.commit()
except Exception as e:
    session.rollback()  # 에러 시 롤백 필수!
    raise

2. 세션 재사용 금지

# 나쁜 예: 전역 세션
global_session = Session()  # 절대 금지!

# 좋은 예: 함수마다 새 세션
def add_user(name):
    session = Session()
    try:
        user = User(name=name)
        session.add(user)
        session.commit()
    finally:
        session.close()

3. 쿼리 최적화 확인

# echo=True로 실제 SQL 확인
engine = create_engine('sqlite:///mydb.db', echo=True)

# 느린 쿼리 찾기
import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

---

🔗 관련 자료

• SQLAlchemy 공식 문서
• SQLAlchemy ORM 튜토리얼
• Python DB-API

---

📚 이전 학습

Day 78: Python과 SQLite 연동 ⭐⭐⭐⭐

어제는 sqlite3 모듈로 파이썬에서 데이터베이스를 다루는 방법을 배웠어요!

📚 다음 학습

Day 80: Phase 8 실전 프로젝트 - 할 일 관리 앱 🎉

내일은 Phase 8에서 배운 모든 내용을 종합하여 완전한 Todo 앱을 만들어봐요!

---

“ORM은 마법이 아니라 도구예요. 언제 쓰고, 언제 쓰지 말아야 할지 아는 것이 진정한 실력이랍니다!” 🎯

Day 79/100

Phase 8 거의 완료!

#100DaysOfPython