에듀테크 프로젝트 대시보드
SQLAlchemy 2.0 — 동기 연동
가장 흔한 구성: 동기 엔진 + 요청마다 세션 + def 라우트.
pip install "sqlalchemy>=2.0" "psycopg[binary]"# 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# 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")# 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)]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# 개발 초기엔 이렇게, 이후엔 Alembic 으로 전환
from app.database import Base, engine
import app.models # 모델 import 필수
Base.metadata.create_all(engine)