251031 [ Day 78 ] - SQL (7)

TaeHyun·2025년 10월 31일

TIL

목록 보기

91/183

시작하며

오늘부로 MySQL과 FastAPI에 대한 파트가 모두 끝이났다.

DB Session

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# DB 연결 정보
# user이름:비밀번호@호스트/DB이름
DATABASE_URL = "mysql+mysqlconnector://user:1234@localhost/codingon_db"

# 엔진 생성
engine = create_engine(DATABASE_URL)

# 세션 생성
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

# 모든 ORM 모델 클래스의 부모(기반) 클래스
Base = declarative_base

user 모델 정의

Integer + primary_key = AUTO_INCREMENT 자동 적용
- 비활성화 필요 시 autoincrement=False 속성 추가

from app.db.session import Base
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import Integer, String, DateTime
from datetime import datetime, UTC

class User(Base):
    # 테이블 이름
    __tablename__ = "users"

    # 테이블 구조 맵핑
    user_id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    password: Mapped[str] = mapped_column(String(255), nullable=False)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))

자동으로 DB 테이블 생성

main.py 또는 따로 관리 파일에 Base.metadata.create_all(bind=engine) 추가

DB 세션 주입

반복 줄여 재사용성 증가
- 매번 DB 세션을 직접 만들고 닫는 코드를 쓰지 않아도 됨
명확한 생명주기 관리
- 커넥션 누수 방지
자동 주입
- FastAPI가 의존성 그래프를 관리해서 DB 세션, 인증된 사용자, 설정 객체 등을 자동으로 함수에 주입

from app.db.session import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

CRUD with ORM

READ

db.query(Model) : DB에서 데이터 조회 (SELECT)

@router.get("/users")
def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

select() : 최신 방식
- from sqlalchemy import select
- 쿼리 구성과 실행을 분리
- 비동기(async) 지원
- where() 처럼 기존 SQL문과 일관성 유지
scalars() : select로 가져온 객체 리스트 반환
- 단일 테이블 조회 시 사용
execute() : Row 객체 반환(튜플 형태)
- 여러 테이블 조회 또는 JOIN 사용 시 사용

@router.get("/posts")
def get_posts(db: Session = Depends(get_db)):
    # select 사용
    stmt = select(Post).order_by(Post.created_at.desc())
    posts = db.scalars(stmt).all()
    return posts

CREATE

db.add(obj) : 새 객체를 DB에 등록 (CREATE)
db.commit() : 변경사항을 DB에 반영
db.rollback() : 변경 사항을 모두 취소

@router.post("/users")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    try:
        new_user = User(username=user.username, password=user.password, name=user.name, email=user.email)

        db.add(new_user)
        db.commit()
        return {"message": "사용자 등록 완료"}
    
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(e))

UPDATE

filter() : 조회 조건 지정 (WHERE)
first() : 조건에 맞는 첫번째 값 반환

@router.put("/users/{user_id}")
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    try:
        db_user = db.query(User).filter(User.user_id == user_id).first()
        if not db_user:
            raise HTTPException(status_code=404, detail="해당 사용자가 없습니다.")
        
        db_user.name = user.name
        db_user.email = user.email
        db.commit()
        return {"message": "사용자 정보 수정 완료"}
        
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(e))

DELETE

db.delete(obj) : 객체를 DB에서 삭제 (DELETE)

@router.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    try:
        db_user = db.query(User).filter(User.user_id == user_id).first()
        if not db_user:
            raise HTTPException(status_code=404, detail="해당 사용자가 없습니다.")

        db.delete(db_user)
        db.commit()
        return {"message": "사용자 삭제 완료"}
    
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(e))