한양과학기술고등학교

에듀테크 프로젝트 대시보드

SQLAlchemy 2.0 — 동기 연동

create_engine sessionmaker DeclarativeBase Mapped get_db

가장 흔한 구성: 동기 엔진 + 요청마다 세션 + def 라우트.

1. 설치

pip install "sqlalchemy>=2.0" "psycopg[binary]"

2. 엔진과 세션 팩토리

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from app.config import get_settings

engine = create_engine(
    get_settings().database_url,      # postgresql+psycopg://...
    pool_size=5, max_overflow=10,
    pool_pre_ping=True,               # 죽은 커넥션 자동 감지
    echo=False,
)

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

class Base(DeclarativeBase):
    pass

3. 모델 정의 (2.0 스타일)

# app/models.py
from datetime import datetime
from sqlalchemy import String, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(default="")
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    author: Mapped[Author] = relationship(back_populates="posts")

4. get_db 의존성

# app/deps.py
from typing import Annotated
from fastapi import Depends
from sqlalchemy.orm import Session
from app.database import SessionLocal

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

DbDep = Annotated[Session, Depends(get_db)]

5. 라우트에서 쿼리 (2.0 select)

from sqlalchemy import select

@app.get("/posts")
def list_posts(db: DbDep, skip: int = 0, limit: int = 20):
    stmt = select(Post).order_by(Post.id.desc()).offset(skip).limit(limit)
    return db.scalars(stmt).all()

@app.get("/posts/{post_id}")
def get_post(post_id: int, db: DbDep):
    post = db.get(Post, post_id)          # PK 조회
    if post is None:
        raise HTTPException(404)
    return post

6. 테이블 생성 (초기/프로토타입)

# 개발 초기엔 이렇게, 이후엔 Alembic 으로 전환
from app.database import Base, engine
import app.models  # 모델 import 필수

Base.metadata.create_all(engine)
"커넥션은 요청마다 열지 말고 풀에서 빌려 쓴다"