1. ORM이란? 🤔

ORM (Object-Relational Mapping) 의 핵심 개념

ORM은 객체-관계 매핑 의 약자로, 파이썬의 객체와 데이터베이스의 테이블을 자동으로 연결해주는 기술입니다!

┌─────────────────────────────────────────────┐
│          🐍 Python 세계 (객체 지향)          │
│                                              │
│  class User:                                │
│      id = 1                                 │
│      name = "홍길동"                         │
│      email = "hong@example.com"             │
└─────────────────────────────────────────────┘
                    ↕️ ORM이 자동 변환
┌─────────────────────────────────────────────┐
│        💾 Database 세계 (관계형 데이터)      │
│                                              │
│  users 테이블:                               │
│  ┌────┬────────┬──────────────────┐        │
│  │ id │ name   │ email            │        │
│  ├────┼────────┼──────────────────┤        │
│  │ 1  │ 홍길동  │ hong@example.com │        │
│  └────┴────────┴──────────────────┘        │
└─────────────────────────────────────────────┘

매핑 관계 이해하기!

💡 쉽게 이해하기

Python에서 user = User(name="홍길동") 이라고 쓰면,

ORM이 자동으로 INSERT INTO users (name) VALUES ('홍길동') SQL을 생성해서 실행해주는거죠!

2. ORM을 쓰는 이유?

Rawl SQL의 문제점

Raw SQL의 문제점이 뭔지부터 알아야 합니다.

1. SQL 문자열을 직접 작성해야 함

cursor.execute("INSERT INTO users (name, email) VALUES ('홍길동', 'hong@example.com')")

2. 😱 타입 실수가 발생하기 쉬움

cursor.execute("INSERT INTO users (age) VALUES ('스물다섯')")  # 에러!

3. 😱 SQL 인젝션 위험

user_input = "'; DROP TABLE users; --"
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")  # 위험!

4. 😱 데이터베이스 변경 시 모든 SQL 수정 필요

• SQLite → PostgreSQL 이동 시 문법 차이로 대규모 수정

ORM의 장점

1. 파이썬 객체로 직관적으로 작업 가능

user = User(name='홍길동', email='hong@example.com')
db.session.add(user)
db.session.commit()

2. 타입 체크 자동!

user.age = "스물다섯"  # IDE가 미리 경고!

3. SQL 인젝션 자동 방지

User.query.filter_by(name=user_input).first()  # 안전!

4. 데이터베이스 변경에 강함

• SQLite -> PostgreSQL로 바꿔도 코드 수정 최소화 가능

3. 실습 환경 준비

이론만 백날 들어도 한 번의 실습만 못합니다!

프로젝트 구조

orm_tutorial/
├── requirements.txt           # 필요한 패키지
├── 1_raw_sql_crud.py         # Raw SQL 방식
├── 2_orm_crud.py             # ORM 방식
├── 3_comparison.py           # 비교 실습
├── raw_sql_database.db       # Raw SQL 데이터베이스
└── orm_database.db           # ORM 데이터베이스

패키지 설치

1) requirements.txt 생성

Flask-SQLAlchemy==3.1.1

4. Raw SQL로 CRUD 구현하기

4-1. 데이터베이스 연결 관리

Rawl SQL 방식은 데이터베이스 연결을 직접 관리해줘야 합니다 ..!

import sqlite3
from datetime import datetime


class DatabaseManager:
    def __init__(self, db_name: str = "test.db"):
        self.db_name = db_name
        self.connection: None | sqlite3.Connection = None
        self.cursor: None | sqlite3.Cursor = None

    def connect(self) -> None:
        """데이터베이스 연결"""
        # SQLite 데이터베이스에 연결
        self.connection = sqlite3.connect(self.db_name)

        # Row 타입으로 결과를 받아 딕셔너리로 처리
        self.connection.row_factory = sqlite3.Row

        # cursor: SQL 명령 실행하는 객체
        self.cursor = self.connection.cursor()

        print(f"✅ 데이터베이스 연결: {self.db_name}")

    def disconnect(self) -> None:
        """데이터베이스 연결 종료"""
        if self.connection:
            self.connection.close()
            print(f"✅ 데이터베이스 연결 종료")

• connection: 데이터베이스와의 연결
• cursor: SQL 명령을 실행하는 도구
• row_factory: 결과를 어떤 형태로 받을지 설정

한 번 실행해볼까요?

if __name__ == "__main__":
    dbm = DatabaseManager()
    dbm.connect()
    dbm.disconnect()

✅ 데이터베이스 연결: test.db
✅ 데이터베이스 연결 종료

4-2. 테이블 생성

def create_table(self) -> None:
    """users 테이블 생성"""

    # users 테이블 생성 쿼리 직접 생성
    create_table_sql = """
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            age INTEGER,
            created_at TEXT NOT NULL
        )
    """
    print(f"\n테이블 생성 SQL:\n{create_table_sql}")

    # SQL 실행
    self.cursor.execute(create_table_sql)

    # 변경사항을 데이터베이스에 저장
    self.connection.commit()

    print("✅ users 테이블 생성 완료!")

⚠️ 주의사항

commit()을 하지 않으면 변경사항이 저장되지 않습니다!

4-3. CREATE - 사용자 생성

def create_user(self, name: str, email: str, age: int | None = None) -> str | None:
    """새로운 사용자 생성 (Create)"""
    print(f"\n사용자 생성: {name}, {email}")

    created_at = datetime.now().isoformat()

    # SQL Insert 문 작성
    # ?는 플레이스홀더 (값이 들어갈 자리)
    insert_sql = """
        INSERT INTO users (name, email, age, created_at)
        VALUES (?, ?, ?, ?)
    """

    print(f"📝 실행 SQL:\n{insert_sql}")
    print(f"📝 값: {(name, email, age, created_at)}")

    try:
        # SQL 실행 - 플레이스홀더에 값을 전달
        self.cursor.execute(insert_sql, (name, email, age, created_at))

        # 변경사항 저장
        self.connection.commit()

        # 방금 인서트된 행의 ID 가져오기
        user_id = self.cursor.lastrowid

        print(f"✅ 사용자 생성 완료! ID: {user_id}")
        return user_id
    except sqlite3.IntegrityError as e:
        print(f"❌ 에러: {str(e)}")
        return None

💡 플레이스홀더 (?) 사용 이유

?를 사용하면 SQL 인젝션을 방지할 수 있습니다.

나쁜 예: f"INSERT INTO users VALUES ('{name}')"

좋은 예: cursor.execute("INSERT INTO users VALUES (?)", (name,))

사용자를 한 번 생성해봅시다.

if __name__ == "__main__":
    dbm = DatabaseManager()

    # DB 연결
    dbm.connect()

    # 테이블 생성
    dbm.create_table()

    # 유저 생성 (Create)
    user_id: str = dbm.create_user(
        name="jh", email="jh@jh.com", age=29
    )

사용자 생성: jh, jh@jh.com
📝 실행 SQL:

            INSERT INTO users (name, email, age, created_at)
            VALUES (?, ?, ?, ?)
        
📝 값: ('jh', 'jh@jh.com', 29, '2025-11-09T08:33:44.539447')

4-4. READ - 사용자 조회

def get_user_by_id(self, user_id: int) -> dict|None:
    """ID로 사용자 조회"""
    print(f"🔍 ID로 사용자 조회: {user_id}")

    select_sql = """
        SELECT id, name, email, age, created_at FROM users WHERE id = ?
    """

    # SQL 실행
    self.cursor.execute(select_sql, (user_id,))

    # 결과 가져오기
    row = self.cursor.fetchone()
    
    if row:
        user = dict(row) # Row를 딕셔너리로 변환
        print(f"사용자 찾음!: {user}")
        return user
    else:
        print(f"사용자 못찾음!")
        return None

💡 fetch 메소드 종류

• fetchone(): 첫 번째 결과만 반환
• fetchall(): 모든 결과를 리스트로 반환
• fetchmany(n): n개의 결과만 반환

위 코드에서 추가로 확인해볼게 뭐가 있을까요?

이전에 DatabaseManager 클래스에서 SQLite3의 connection 객체에 다음과 같은 속성을 넣었던 것이 기억나십니까?

# Row 타입으로 결과를 받아 딕셔너리로 처리
self.connection.row_factory = sqlite3.Row

여기서 위 속성을 주석 처리하고 get_user_by_id 메소드에서 row 데이터를 조회해보면 다음과 같이 데이터가 넘어옵니다.

(1, 'jhlee37', 'jhlee37@cafe24corp.com', 29, '2025-11-09T08:00:46.327799')

데이터를 조회해봤을 때 넘어온 데이터가 어떤 컬럼의 데이터인지 알아보기가 쉽지 않죠?

그럼 다시 주석 처리 했던 self.connection.row_factory = sqlite3.Row를 주석 해제하고 다시 실행해보면?

{'id': 1, 'name': 'jhlee37', 'email': 'jhlee37@cafe24corp.com', 'age': 29, 'created_at': '2025-11-09T08:00:46.327799'}

이런 식으로 row 데이터가 보기 좋게 dict 형태로 처리되는 것을 볼 수 있습니다!

모든 사용자 조회하기

위 예시는 단일 사용자만 조회할 수 있었죠?

전체 사용자를 조회하고 싶다면, fetchall()로 쿼리에 대한 모든 결과를 리스트로 반환하여 가져오면 됩니다.

def get_all_users(self) -> list[dict]:
    """모든 사용자 조회"""
    print("\n📋모든 사용자 조회")

    select_sql = """
    SELECT id, name, email, age, created_at FROM users ORDER BY id desc
    """
    self.cursor.execute(select_sql)

    # 모든 결과 행 가져오기
    rows = self.cursor.fetchall()

    # 각 행을 딕셔너리로 전환
    users = [dict(row) for row in rows]

    print(f"✅ {len(users)} 명의 유저를 찾음.")
    return users

4-5. UPDATE - 사용자 수정

def update_user(
    self, user_id: int, name: str = None, email: str = None, age: int = None
) -> bool:
    """사용자 정보 수정"""

    print(f"✏️  사용자 수정: ID {user_id}")

    # 수정할 필드 목록
    update_fields = []
    update_values = []

    # 전달된 값만 업데이트 목록에 추가
    if name is not None:
        update_fields.append("name = ?")
        update_values.append(name)

    if email is not None:
        update_fields.append("email = ?")
        update_values.append(email)

    if age is not None:
        update_fields.append("age = ?")
        update_values.append(age)

    # 수정할 내용이 없는 경우
    if not update_fields:
        print("⚠️ 수정할 내용이 없습니다.")
        return False

    # WHERE 절의 user_id도 추가
    update_values.append(user_id)

    # SQL Update 문 동적 생성
    update_sql = f"""
        UPDATE users
        SET {", ".join(update_fields)}
        WHERE id = ?
    """

    print(f"실행할 SQL문!\n{update_sql}")
    print(f"값: {tuple(update_values)}")
    try:
        self.cursor.execute(update_sql, tuple(update_values))
        self.connection.commit()

        # 실제로 수정된 행의 개수
        if self.cursor.rowcount > 0:
            print(f"✅ 사용자 수정 완료! (업데이트된 행: {self.cursor.rowcount})")
            return True
        else:
            print(f"❌ 해당 ID의 사용자가 없음!")
            return False

    except sqlite3.IntegrityError as e:
        print(f"에러: {str(e)}")
        return False

id가 4번인 사용자를 수정해보겠습니다.

if __name__ == "__main__":
    # ...    
    # 유저 정보 업데이트
    dbm.update_user(user_id=4, name="up_jhlee37", email="up_jhlee37@com", age=30)
    dbm.get_user_by_id(user_id=4)  # 수정됐는지 확인

출력 값을 확인해볼까요?

✅ 데이터베이스 연결: test.db
✅ users 테이블 생성 완료!
✏️  사용자 수정: ID 4
실행할 SQL문!

            UPDATE users
            SET name = ?, email = ?, age = ?
            WHERE id = ?
        
값: ('up_jhlee37', 'up_jhlee37@com', 30, 4)
✅ 사용자 수정 완료! (업데이트된 행: 1)
🔍 ID로 사용자 조회: 4
유저 찾음!: {'id': 4, 'name': 'up_jhlee37', 'email': 'up_jhlee37@com', 'age': 30, 'created_at': '2025-11-15T07:25:41.487555'}
✅ 데이터베이스 연결 종료

정상적으로 id가 4번인 유저의 데이터가 수정된 것을 볼 수 있습니다!

4-6. DELETE - 사용자 삭제

마지막으로 사용자를 삭제하는 함수를 작성해보며, Raw SQL로 CRUD를 구현하는 챕터를 마무리해보겠습니다.

def delete_user(self, user_id: int) -> bool:
    """사용자 삭제"""
    print(f"\n🗑️ 사용자 삭제: ID {user_id}")

    delete_sql = """
    DELETE FROM users
    WHERE id = ?
    """

    self.cursor.execute(delete_sql, (user_id,))
    self.connection.commit()

    if self.cursor.rowcount > 0:
        print(f"✅ 사용자 삭제 완료! (삭제된 행: {self.cursor.rowcount})")
        return True
    else:
        print(f"❌ 해당 ID의 사용자가 없음!")
        return False

id가 3번인 사용자를 삭제해보겠습니다.

if __name__ == "__main__":
    # ...    
    dbm.delete_user(user_id=3)

    # 유저가 삭제됐는지 확인하기.
    users = dbm.get_all_users()
    print(users)

출력 값을 확인해봅시다.

✅ 데이터베이스 연결: test.db
✅ users 테이블 생성 완료!

🗑️ 사용자 삭제: ID 3
✅ 사용자 삭제 완료! (삭제된 행: 1)

📋모든 사용자 조회
✅ 1 명의 유저를 찾음.
[{'id': 2, 'name': 'jhlee38', 'email': 'jhlee38@cafe24corp.com', 'age': 29, 'created_at': '2025-11-09T08:02:06.148338'}]
✅ 데이터베이스 연결 종료

id가 3번인 사용자가 삭제되어, 더이상 조회되지 않는 것을 볼 수 있습니다.

5. ORM으로 CRUD 구현하기

이제 Raw SQL로 구현한 모든 기능을 ORM으로 구현해볼 겁니다.

코드가 얼마나 간결해지는지 한 번 비교해봅시다.

5-0. 모듈 불러오기

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

현재 불러온 함수와 클래스에 대한 설명은 다음과 같습니다.

5-1. 데이터베이스 연결하기

데이터베이스의 경우 이전에는 sqlite3를 사용하였으나,

이번 실습에서는 MySQL로 전환하여 실습을 진행해보도록 하겠습니다.

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

# 타입힌트 처리용 모듈
import sqlalchemy.orm as orm
import sqlalchemy.exc as exc
import sqlalchemy.engine as eng

# Base 클래스 생성
# 모든 ORM은 이 Base를 상속받음.
Base: orm.decl_api.DeclarativeMeta = declarative_base()

# MySQL 연결 정보
DB_USER: str = "root"
DB_PASS: str = "custom"
DB_HOST: str = "192.168.219.110"
DB_PORT: int = 3333
DB_NAME: str = "orm"

# 데이터베이스 엔진 생성
try:
    engine: eng.base.Engine = create_engine(
        f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}",
        echo=True,  # SQL 쿼리문을 자동으로 출력해줌 (학습용)
    )

except exc.ArgumentError as e:
    print(f"[🔴] Argument Error!\nReason: {str(e)}")

# Session 클래스 생성
# Session: 데이터베이스와의 대화 (conversation)
SessionLocal: orm.session.sessionmaker = sessionmaker(bind=engine)

5-2. 클래스(테이블) 생성하기

ORM 모델 정의하기 (테이블 스키마 정의)

# ------------------ ORM 모델 정의 ------------------
class User(Base):
    """
    User 테이블을 나타내는 ORM 모델

    User 클래스:
    - 테이블 구조를 정의
    - 각 인스턴스가 테이블의 한 행(row)을 나타냄
    - Python 객체처럼 다룰 수 있음.
    """

    # 테이블 이름 지정
    __tablename__ = "users"

    # 컬럼 정의
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), nullable=False, unique=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)

    def __repr__(self) -> str:
        """객체를 문자열로 표현함"""
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

    def to_dict(self) -> dict:
        """객체를 딕셔너리로 변환"""
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
            "age": self.age,
            "created_at": self.created_at,
        }

💡 ORM 모델의 구조

• __tablename__: 테이블 이름
• Column(): 각 컬럼 정의
• primary_key=True: 기본 키 설정
• nullable=False: NOT NULL 제약
• unique=True: UNIQUE 제약
• default: 기본값 설정

테이블 생성

# ------------------ 테이블 생성------------------
def create_tables() -> None:
    """정의된 모든 모델의 테이블 생성"""
    print("\n📝 테이블 생성 중 ...")
    # Base.metadata.create_all():
    # - Base를 상속받은 모든 클래스를 찾음
    # - 각 클래스에 대응하는 테이블을 생성
    # - 이미 존재하는 테이블은 건드리지 않음
    Base.metadata.create_all(bind=engine)
    print("✅ 테이블 생성 완료!")

5-3. CREATE - 사용자 생성

# ------------------ CRUD ------------------
class UserCRUD:
    """ORM을 사용한 User CRUD 작업"""

    def __init__(self):
        self.session: orm.session.sessionmaker = SessionLocal()

    def create_user(self, name: str, email: str, age: int = None) -> None | User:
        """
        새로운 사용자 생성

        ORM 방식:
        1. User 객체 생성 (메모리에만 존재)
        2. session.add()로 세션에 추가
        3. session.commit()으로 DB에 실제 저장
        """

        print(f"\n➕ 사용자 생성: {name}, {email}")

        # 1. User 객체 생성
        # 이 시점에서는 메모리에만 존재
        new_user = User(name=name, email=email, age=age)

        print(f"생성된 객체: {new_user}")
        print(f"ID: {new_user.id}")  # None (아직 DB에 없음!)

        try:
            # 2. Session에 추가
            self.session.add(new_user)

            # 3. DB에 실제 저장
            # INSERT SQL이 자동 생성되어 실행됨
            self.session.commit()

            # commit 후에는 ID가 생성됨!
            print(f"사용자 생성 완료! ID: {new_user.id}")
            return new_user

        except Exception as e:
            print(f"[🔴] {str(e)}")
            return None

💡 ORM 생성 프로세스

1. 객체 생성: user = User(name="홍길동") → 메모리에만 존재
2. 세션 추가: session.add(user) → “저장할 예정” 표시
3. 커밋: session.commit() → 실제 DB에 저장

한 번 User 객체를 만들어서 사용자를 생성해보도록 합시다.

if __name__ == "__main__":
    """ORM CRUD 실습"""
    print("=" * 80)
    print("SQLAlchemy ORM을 사용한 CRUD 실습")
    print("=" * 80)

    # 1. 테이블 생성
    create_tables()

    # 2. CRUD 객체 생성
    user_crud = UserCRUD()

    # 3. 새로운 유저 생성
    new_user: User | None = user_crud.create_user(
        name="jhlee04", email="jhlee04@jhlee04", age=29
    )
    print(
        f"새로운 유저 정보: ID: {new_user.id}, NAME: {new_user.name}, EMAIL: {new_user.email}"
    )

출력 값을 한 번 보도록 할게요.

================================================================================
SQLAlchemy ORM을 사용한 CRUD 실습
================================================================================

📝 테이블 생성 중 ...
2025-11-15 10:09:51,140 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2025-11-15 10:09:51,140 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-11-15 10:09:51,141 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2025-11-15 10:09:51,141 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-11-15 10:09:51,141 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2025-11-15 10:09:51,142 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-11-15 10:09:51,142 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-11-15 10:09:51,142 INFO sqlalchemy.engine.Engine DESCRIBE `orm`.`users`
2025-11-15 10:09:51,142 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-11-15 10:09:51,144 INFO sqlalchemy.engine.Engine COMMIT
✅ 테이블 생성 완료!

➕ 사용자 생성: jhlee04, jhlee04@jhlee04
생성된 객체: <User(id=None, name='jhlee04', email='jhlee04@jhlee04')>
ID: None
2025-11-15 10:09:51,145 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-11-15 10:09:51,146 INFO sqlalchemy.engine.Engine INSERT INTO users (name, email, age, created_at) VALUES (%(name)s, %(email)s, %(age)s, %(created_at)s)
2025-11-15 10:09:51,146 INFO sqlalchemy.engine.Engine [generated in 0.00013s] {'name': 'jhlee04', 'email': 'jhlee04@jhlee04', 'age': 29, 'created_at': datetime.datetime(2025, 11, 15, 10, 9, 51, 146782)}
2025-11-15 10:09:51,147 INFO sqlalchemy.engine.Engine COMMIT
2025-11-15 10:09:51,150 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-11-15 10:09:51,152 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.age AS users_age, users.created_at AS users_created_at 
FROM users 
WHERE users.id = %(pk_1)s
2025-11-15 10:09:51,152 INFO sqlalchemy.engine.Engine [generated in 0.00010s] {'pk_1': 12}
사용자 생성 완료! ID: 12
새로운 유저 정보: ID: 12, NAME: jhlee04, EMAIL: jhlee04@jhlee04

여러 출력문이 뜨고 있죠?

SQL 관련 로그가 뜨는 것은 위에서 설정한 아래 코드 때문입니다.

engine: eng.base.Engine = create_engine(
    f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}",
    echo=True,  # SQL 쿼리문을 자동으로 출력해줌 (학습용)
)

echo 속성을 True로 줬기 때문에 SQL 쿼리문에 대한 로그가 뜨고 있는 것을 볼 수 있어요.

Raw SQL과 비교

Raw SQL을 사용했을 때와는 어떤 점이 다른지 코드로 비교해볼까요?

# Raw SQL: 19줄
insert_sql = """
INSERT INTO users (name, email, age, created_at)
VALUES (?, ?, ?, ?)
"""
created_at = datetime.now().isoformat()
try:
    cursor.execute(insert_sql, (name, email, age, created_at))
    connection.commit()
    user_id = cursor.lastrowid
    return user_id
except sqlite3.IntegrityError as e:
    print(f"에러: {e}")
    return None

# ORM: 7줄
try:
    user = User(name=name, email=email, age=age)
    session.add(user)
    session.commit()
    return user
except Exception as e:
    session.rollback()
    return None

5-4. READ - 사용자 조회

id 컬럼 값 기반 단일 사용자 조회

def get_user_by_id(self, user_id: int) -> User | None:
    """ID로 사용자 조회"""
    print(f"\n🔍 ID로 사용자 조회: {user_id}")
    # get(): 기본 키로 조회
    # SELECT * FROM users WHERE id = ?와 동일
    user: User = self.session.get(User, user_id)
    if user:
        print(f"사용자 찾음: {user.to_dict()}")
    else:
        print("사용자를 찾을 수 없음!")
    return user

id가 1인 사용자를 조회해보겠습니다.

if __name__ == "__main__":
    """ORM CRUD 실습"""
    # ...
    # 4. 단일 유저 조회 (user_id, READ)
    user: User = user_crud.get_user_by_id(user_id=1)
    print(user.to_dict())

출력 값을 조회해볼게요.

사용자 찾음: {'id': 1, 'name': 'jhlee01', 'email': 'jhlee01@jhlee01', 'age': 29, 'created_at': datetime.datetime(2025, 11, 15, 9, 28, 18)}
{'id': 1, 'name': 'jhlee01', 'email': 'jhlee01@jhlee01', 'age': 29, 'created_at': datetime.datetime(2025, 11, 15, 9, 28, 18)}

id가 1번인 사용자 데이터가 조회됩니다.

to_dict() 메소드를 사용하여 객체의 정보가 딕셔너리로도 정상적으로 조회되는 것을 볼 수 있어요.

모든 사용자 조회

def get_all_users(self) -> list[User]:
    """모든 사용자 조회"""
    print("\n📋 모든 사용자 조회")

    # query(User): User 테이블에 대한 쿼리
    # order_by(User.id): ORDER BY id
    # all(): 모든 결과를 리스트로 반환
    users = self.session.query(User).order_by(User.id).all()
    print(f"{len(users)} 명의 사용자 찾음.")
    return users

모든 사용자 정보를 list로 리턴하도록 작성하였습니다.

모든 사용자를 조회해봅시다.

if __name__ == "__main__":
    """ORM CRUD 실습"""
    # ...
    # 5. 모든 유저 조회
    users: list = user_crud.get_all_users()
    print(users)

4 명의 사용자 찾음.
[<User(id=1, name='jhlee01', email='jhlee01@jhlee01')>, <User(id=9, name='jhlee02', email='jhlee02@jhlee02')>, <User(id=10, name='jhlee03', email='jhlee03@jhlee03')>, <User(id=12, name='jhlee04', email='jhlee04@jhlee04')>]

이름으로 검색

def get_users_by_name(self, name_keyword: str) -> list[User]:
    """유저 이름으로 조회"""
    print("\n🔍 이름으로 검색")

    # filter(): WHERE 절
    # like(): SQL의 LIKE 연산자
    users = (
        self.session.query(User)
        .filter(User.name.like(f"%{name_keyword}%"))
        .order_by(User.name)
    ).all()
    print(f"키워드: '{name_keyword}' - {len(users)}명의 사용자 찾음.")
    return users

sessionmaker 객체가 가지고 있는 filter 메소드를 통해서 SQL의 LIKE 연산자도 처리할 수 있습니다.

jh 라는 문자열을 포함하는 유저들을 조회해보겠습니다.

if __name__ == "__main__":
    """ORM CRUD 실습"""
    # ...
    # 6. 이름으로 검색 (LIKE 연산자)
    users = user_crud.get_users_by_name(name_keyword="jh")

키워드: jh - 4명의 사용자 찾음.
[<User(id=1, name='jhlee01', email='jhlee01@jhlee01')>, <User(id=9, name='jhlee02', email='jhlee02@jhlee02')>, <User(id=10, name='jhlee03', email='jhlee03@jhlee03')>, <User(id=12, name='jhlee04', email='jhlee04@jhlee04')>]

Raw SQL과 비교

# ---- Raw SQL: 복잡한 문자열 조작
select_sql = """
SELECT id, name, email, age, created_at
FROM users
WHERE name LIKE ?
ORDER BY name ASC
"""
cursor.execute(select_sql, (f"%{name_keyword}%",))
rows = cursor.fetchall()
users = [dict(row) for row in rows]

# ---- ORM: 메소드 체이닝으로 직관적
users = session.query(User)\
    .filter(User.name.like(f"%{name_keyword}%"))\
    .order_by(User.name)\
    .all()

5-5. UPDATE - 사용자 수정

id 값을 기반으로 사용자 정보를 수정할 때!

def update_user(
    self, user_id: int, name: str = None, email: str = None, age: int = None
) -> None | User:
    """
    사용자 정보 수정

    ORM 방식:
    1. 객체 조회
    2. 객체의 속성 수정
    3. commit()으로 변경사항 저장
    """

    print(f"\n ✏️ 사용자 수정: ID: {user_id}")

    # 1. 사용자 조회
    user: User = self.session.get(User, user_id)
    if not user:
        print(f"해당 ID({user_id})의 사용자가 없습니다")
        return None

    # 수정 전의 사용자 정보 출력
    print(f"    수정 전: {user.to_dict()}")

    # 2. 속성 수정
    # 객체의 속성을 직접 수정
    # ORM이 변경사항을 자동으로 추적함.
    if name is not None:
        user.name = name
        print(f"    이름 변경: {user.name}")

    if email is not None:
        user.email = email
        print(f"    이메일 변경: {user.email}")

    if age is not None:
        user.age = age
        print(f"    나이 변경: {user.age}")

    try:
        # 3. 변경사항 적용
        # UPDATE SQL이 자동 생성됨.
        self.session.commit()

        print(f"✅ 사용자 수정 완료!")
        print(f"    수정 후: {user.to_dict()}")

        return user

    except Exception as e:
        self.session.rollback()  # 이전 설정으로 되돌리기
        print(f"에러: {str(e)}")
        return None

id 값이 1번인 사용자의 정보를 수정해볼까요?

if __name__ == "__main__":
    """ORM CRUD 실습"""
    # ...    
    # 7. 사용자 업데이트
    update_user: User = user_crud.update_user(user_id=1, name="jhlee_updated_01")
    read_user: User = user_crud.get_user_by_id(user_id=1)

위 코드를 실행하고 id가 1번인 유저의 name 필드 값이 jhlee_updated_01로 변경되어 있으면 성공이네요?

🔍 ID로 사용자 조회: 1
사용자 찾음: {'id': 1, 'name': 'jhlee_updated_01', 'email': 'jhlee01@jhlee01', 'age': 29, 'created_at': datetime.datetime(2025, 11, 15, 9, 28, 18)}

정상적으로 id가 1번인 사용자의 name 값이 변경되었습니다!

Raw SQL과 비교

# Raw SQL: 복잡한 동적 SQL 생성
update_fields = []
update_values = []
if name is not None:
    update_fields.append("name = ?")
    update_values.append(name)
if email is not None:
    update_fields.append("email = ?")
    update_values.append(email)
update_values.append(user_id)
update_sql = f"UPDATE users SET {', '.join(update_fields)} WHERE id = ?"
cursor.execute(update_sql, tuple(update_values))
connection.commit()

# ORM: 직관적인 속성 수정
user = session.get(User, user_id)
if name is not None:
    user.name = name
if email is not None:
    user.email = email
session.commit()

Raw SQL의 경우 직접 쿼리문을 작성하여 실행하다보니 코드가 길어지고 가독성이 상당히 떨어집니다 …

5-6. DELETE - 사용자 삭제

def delete_user(self, user_id: int) -> bool:
    """
    사용자 삭제

    ORM 방식:
    1. 객체 조회
    2. session.delete()로 삭제 표시
    3. commit()으로 실제 삭제
    """

    print(f"\n🗑️ 사용자 삭제: {user_id}")

    # 1. 사용자 조회
    user: User = self.session.get(User, user_id)
    if not user:
        print(f"해당 ID({user_id})의 사용자가 없습니다!")
        return False

    # 2. 삭제 표시
    self.session.delete(user)

    # 3. 실제 삭제
    # DELETE SQL이 자동 생성되어 실행됨.
    self.session.commit()
    return True

id가 1번인 사용자를 삭제해본 후, 실제로 삭제됐는지 조회해봅시다.

if __name__ == "__main__":
    """ORM CRUD 실습"""
    # ...
    # 9. 사용자 삭제
    user_delete: bool = user_crud.delete_user(user_id=1)
    user_crud.get_user_by_id(user_id=1)  # 실제 삭제되었는지 id 기반 조회

사용자를 찾을 수 없음!

id가 1번인 사용자가 삭제되어 사용자를 찾을 수 없음! 이라는 문구가 출력되네요!

마무리하며

지금까지 Raw SQL과 ORM을 각각 비교하며 실습을 진행해봤습니다.

확실히 일반적인 CRUD에 있어서는 ORM을 사용하는 것이 작업 효율성을 꽤 높일 수 있을 것 같다는 생각이 들었습니다.

다만 복잡한 쿼리나 성능에 있어서 중요한 로직 등은 Raw SQL과 병행하며 코드를 작성하는 것이 좋아보입니다!

오늘 배운 ORM 내용을 기반으로

추후에 간단한 도서 관리 API 를 Flask 프레임워크 기반으로 구현해보려고 합니다!

다음에 만나요!