Session 관리 (SQLAlchemy)

DB 작업이 늘어나면 “Session을 어떻게 만들고, 어디까지 공유하고, 언제 닫을지”가 곧 장애 포인트가 된다.
이 글은 실무에서 자주 쓰는 Session 관리 패턴을 예제/실습 중심으로 정리한다. ✅

---

오늘 목표

• Session이 무엇인지 “대화 창구” 관점에서 이해
• 매번 Session(engine)로 만들지 말고 sessionmaker로 중앙화
• get_db 패턴(Generator)으로 생성/정리를 자동화
• 트랜잭션을 commit/rollback 또는 session.begin()으로 안전하게 관리
• 실무에서 많이 터지는 함정(지연 로딩, 커밋 후 만료, 연결 고갈)을 예방

---

Session이란?

Session은 애플리케이션과 DB 사이의 “대화 창구”다.
단순히 쿼리 실행만 하는 게 아니라 트랜잭션 범위, 객체 상태, 연결(커넥션) 관리를 함께 맡는다.

Session 관리가 망가지면 생기는 대표 증상

상황	문제
요청마다 여기저기서 Session 생성	설정 분산 + 중복 코드 + 유지보수 지옥
Session을 안 닫음	커넥션 풀 고갈 → 서버가 멈춘 것처럼 보임
같은 트랜잭션으로 묶여야 하는데 Session이 분리됨	데이터 일관성 깨짐 (부분 성공/부분 실패)
테스트에서 DB Session 교체가 어려움	테스트 격리 실패 + flaky test

그림으로 이해

[App Code]  --(Session)-->  [DB Connection Pool]  -->  [Database]
   |                |
   |                +-- 커넥션을 "빌려오고/반납"함
   |
   +-- 객체 상태 관리 (new / dirty / deleted / persistent / detached)

---

sessionmaker로 Session 생성 “중앙화”

매번 Session(engine)를 직접 만들면, 프로젝트가 커질수록 Session 생성 코드가 퍼진다.
실무에서는 sessionmaker로 “Session 공장”을 만들고, 필요한 곳에서 찍어쓴다.

기본 구성

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+pymysql://root:1234@localhost:3306/mydb"

engine = create_engine(
DATABASE_URL,
echo=False,            # 개발 중에는 True로 켜서 SQL 로그 확인 가능
pool_pre_ping=True,    # 오래된 커넥션이 죽었는지 사전 체크 (실무에서 꽤 유용)
)

# Session 공장

SessionLocal = sessionmaker(
bind=engine,
autocommit=False,      # True는 추천하지 않음 (명시적 commit이 안전)
autoflush=True,        # 쿼리 전에 변경사항 flush (대부분 켜둠)
expire_on_commit=True  # 커밋 후 객체를 만료(expire)시켜 "DB 최신값"을 보장 (기본값)
)

# 사용 예시

db = SessionLocal()
try:
# ... DB 작업 ...
pass
finally:
db.close()

프로젝트 구조 예시

project/
  app/
    database.py      # engine, SessionLocal, Base, get_db
    models.py        # ORM 모델들
    main.py          # FastAPI/서버 엔트리

# app/database.py
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = os.getenv("DATABASE_URL", "mysql+pymysql://root:1234@localhost:3306/mydb")

engine = create_engine(
    DATABASE_URL,
    echo=False,
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=True,
    expire_on_commit=True,
)

Base = declarative_base()

def init_db():
    # 데모/학습용. 운영에서는 마이그레이션(Alembic)을 쓰는 게 보통
    Base.metadata.create_all(bind=engine)
    

---

get_db 패턴 (Generator) — “자동 정리”의 표준

실무에서 가장 자주 쓰는 패턴 중 하나가 get_db다.
핵심은 yield로 Session을 넘기고, finally에서 무조건 닫는 것.

왜 Generator가 좋은가?

• 예외가 터져도 finally가 실행되어 close 보장
• 사용자(호출자)는 Session 생명주기를 신경 안 써도 됨

# app/database.py (추가)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

그림으로 이해

get_db() 호출
  |
  +-- db = SessionLocal()         (세션 생성)
  |
  +-- yield db                    (세션을 호출자에게 전달하고 "대기")
  |
  +-- 호출자 작업 끝/예외 발생
  |
  +-- finally: db.close()         (세션 정리 보장)

---

4) FastAPI에서 get_db + Depends로 주입

FastAPI를 쓴다면 get_db는 사실상 표준 패턴이다.
“요청 단위”로 Session을 만들고, 요청이 끝나면 자동으로 닫힌다.

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

from app.database import get_db, init_db
from app.models import Member

app = FastAPI()

@app.on_event("startup")
def on_startup():
    init_db()

@app.get("/members/{member_id}")
def get_member(member_id: int, db: Session = Depends(get_db)):
    member = db.get(Member, member_id)
    if not member:
        return {"ok": False, "error": "NOT_FOUND"}
    return {"ok": True, "id": member.member_id, "name": member.name}

@app.post("/members")
def create_member(name: str, email: str, db: Session = Depends(get_db)):
    member = Member(name=name, email=email)
    db.add(member)
    db.commit()
    db.refresh(member)  # id 같은 DB 생성값을 즉시 쓰려면 refresh가 안전
    return {"ok": True, "id": member.member_id}

Depends(get_db)가 해주는 일은 아래와 같다.

1) get_db() 실행 → Session 생성
2) yield된 Session을 db 파라미터로 주입
3) 라우터 함수 종료
4) finally 실행 → Session close

---

FastAPI 없이도 get_db를 깔끔하게 쓰는 법 (contextmanager)

CLI 스크립트, 배치, 크론, 단순 파이썬 앱에서는 with 문이 편하다.

from contextlib import contextmanager
from app.database import SessionLocal
from app.models import Member

@contextmanager
def get_db_context():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def some_business_logic():
    with get_db_context() as db:
        members = db.query(Member).all()
        for m in members:
            print(m.member_id, m.name)

---

트랜잭션 관리 패턴

기본 commit/rollback 패턴

“둘 다 성공해야 하는 작업”은 하나의 트랜잭션으로 묶는다. 실패하면 rollback.

def transfer_money(db, from_id: int, to_id: int, amount: int):
    try:
        from_account = db.get(Account, from_id)
        to_account = db.get(Account, to_id)

        if from_account.balance < amount:
            raise ValueError("잔액 부족")

        from_account.balance -= amount
        to_account.balance += amount

        db.commit()
    except Exception:
        db.rollback()
        raise

session.begin() (SQLAlchemy 2.0 스타일) — 더 명확한 트랜잭션 블록

begin 블록을 쓰면 “성공 시 자동 commit / 예외 시 자동 rollback”이 된다.

from sqlalchemy.orm import Session
from app.database import engine

with Session(engine) as session:
    with session.begin():
        session.add(Member(name="임제프", email="jeff@gmail.com"))
        session.add(Member(name="김철수", email="chulsoo@gmail.com"))
        # 여기서 예외가 나면 자동 rollback
    # 여기까지 오면 자동 commit

---

팁 (사고가 많이 나는 지점) 🔥

Session 범위 밖에서 “지연 로딩” 접근하면 터진다

ORM 관계(relationship)를 지연 로딩(lazy loading)으로 두면, 속성 접근 순간에 추가 쿼리가 나간다.
그런데 Session이 이미 닫혀 있으면? 추가 쿼리를 못 날려서 에러가 난다.

# 잘못된 예: Session이 닫힌 다음에 관계 속성 접근
def get_member_name(member_id: int str:
    with get_db_context() as db:
        member = db.get(Member, member_id)
    # 여기서 member.team.name 같은 관계 접근을 하면:
    # DetachedInstanceError (세션이 닫힘)
    return member.name

해결 방법: 필요한 관계 데이터는 Session 범위 안에서 접근해서 “값으로” 꺼내놓는다.

def get_member_with_team(member_id: int) dict:
    with get_db_context() as db:
        member = db.get(Member, member_id)
        if not member:
            return {"ok": False, "error": "NOT_FOUND"}

        team_name = None
        if member.team:
            team_name = member.team.name  # Session 범위 내에서 접근

        return {"ok": True, "name": member.name, "team_name": team_name}

커밋 후 객체 만료(expire_on_commit)와 refresh

SQLAlchemy는 기본적으로 commit 이후 객체를 만료(expire)시키는 성향이 있다.
이유는 “DB에서 트리거/기본값/auto_increment로 값이 바뀌었을 수 있으니, 최신 상태를 DB에서 다시 보라”는 철학 때문이다.

상황	무슨 일이 일어남	대응
commit 직후	객체가 expire되어 속성 접근 시 재조회가 발생할 수 있음	즉시 값이 필요하면 refresh
ID/created_at 같은 DB 생성값 필요	python 객체에 아직 없을 수 있음	db.refresh(obj)

def create_member(db, name: str, email: str):
    member = Member(name=name, email=email)
    db.add(member)
    db.commit()

    # DB가 생성한 값(id 등)을 즉시 사용하려면 refresh가 안전
    db.refresh(member)
    return member.member_id

“Session = 쓰레드 안전”이 아니다

Session은 보통 요청 단위/작업 단위로 생성해야 한다.
전역 Session 하나를 여러 요청/쓰레드에서 공유하면 데이터가 섞이고, 트랜잭션 경계도 무너진다.

권장 범위:
  - 웹 서버: 요청(Request) 단위 Session
  - 배치/스크립트: 작업(Job) 단위 Session
금지 패턴:
  - 전역 Session을 만들어서 모든 곳에서 공유

커넥션 풀 고갈을 막는 체크리스트

• finally/with로 close 보장
• 예외 시 rollback
• 긴 작업(대용량 처리)은 “chunk”로 나누고 주기적으로 commit/close
• pool_pre_ping=True로 죽은 커넥션을 사전 감지 (환경에 따라 도움)

---

실습 — “올바른 Session/트랜잭션 패턴” 익히기 🧪

실습 1) get_db_context로 생성/조회/수정/삭제

from app.database import init_db
from app.models import Member

def lab_crud():
init_db()
# CREATE
with get_db_context() as db:
    m1 = Member(name="신짱구", email="zz9@gmail.com")
    db.add(m1)
    db.commit()
    db.refresh(m1)
    member_id = m1.member_id
    print("[CREATE]", member_id)

# READ
with get_db_context() as db:
    m = db.get(Member, member_id)
    print("[READ]", m.member_id, m.name, m.email)

# UPDATE
with get_db_context() as db:
    m = db.get(Member, member_id)
    m.name = "신짱구(수정)"
    db.commit()
    print("[UPDATE]", m.member_id, m.name)

# DELETE
with get_db_context() as db:
    m = db.get(Member, member_id)
    db.delete(m)
    db.commit()
    print("[DELETE]", member_id)

실습 2) 트랜잭션 실패 시 rollback 확인

아래 코드는 중간에 예외를 일부러 발생시켜 “부분 저장”이 일어나지 않는지 확인한다.

def lab_transaction_rollback():
    with get_db_context() as db:
        try:
            db.add(Member(name="A", email="a@gmail.com"))
            db.add(Member(name="B", email="b@gmail.com"))

            # 일부러 터뜨림
            raise RuntimeError("일부러 실패")

            db.commit()
        except Exception as e:
            db.rollback()
            print("[ROLLBACK]", str(e))

    # 실제로 저장이 안 되었는지 확인
    with get_db_context() as db:
        rows = db.query(Member).filter(Member.email.in_(["a@gmail.com", "b@gmail.com"])).all()
        print("[CHECK] saved count =", len(rows))

---

핵심 요약

키워드	한 줄 요약
sessionmaker	Session 생성 설정을 중앙화하는 “공장”
get_db (Generator)	yield + finally로 Session close를 자동 보장
Depends (FastAPI)	요청 단위 Session 생성/정리를 프레임워크가 처리
트랜잭션	commit/rollback 또는 session.begin()으로 일관성 확보
Session 범위	지연 로딩/관계 접근은 Session 살아있을 때 끝내야 안전
refresh	commit 후 DB 생성값(id 등)을 즉시 쓰려면 refresh