SQLAlchemy DB 처리 방식

1. SQLAlchemy + Pydantic 방식

• Pydantic과 함께 쓰면 API/DB 역할 분리가 명확함
• DB 데이터를 "객체"로 다룰 수 있음 (User, Post 등)
• relationship, ForeignKey로 테이블 간 관계 표현 가능

# 1. Pydantic 모델 (입출력용)
class UserCreate(BaseModel):
    username: str
    email: str

# 2. SQLAlchemy ORM 모델 (DB 테이블 대응)
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String)
    email = Column(String)

# 3. ORM 방식으로 DB에 객체 저장
def create_user(db: Session, user_data: UserCreate):
    db_user = User(**user_data.dict())  # 딕셔너리 언팩 → ORM 객체 생성
    db.add(db_user)                     # 세션에 추가
    db.commit()                         # DB에 반영
    db.refresh(db_user)                # 새로 생성된 id 등 갱신
    return db_user

---

• join, filter, subquery, order_by 등 복잡한 조건

from sqlalchemy.orm import aliased
from sqlalchemy import func, desc

def get_top_users(db: Session):
    # 최근 등록된 유저 중 이메일 있는 애들만, 이름 내림차순
    return db.query(User).filter(
        User.email != None,
        User.username.like("gildong%")
    ).order_by(desc(User.id)).limit(10).all()

---

• func, group_by, having 등 집계함수 활용

from sqlalchemy import func

def get_user_count_by_email_domain(db: Session):
    return db.query(
        func.substr(User.email, func.instr(User.email, '@') + 1).label("domain"),
        func.count(User.id).label("count")
    ).group_by("domain").having(func.count(User.id) > 5).all()

---

• 쿼리 힌트 사용 (with_hint, prefix_with, FOR UPDATE 등)

def get_user_with_hint(db: Session):
    # MySQL 인덱스 힌트 사용 예시 (DB 종류에 따라 다름)
    query = db.query(User).with_hint(User, "USE INDEX (idx_username)", dialect_name="mysql")
    return query.filter(User.username == "gildong").first()

---

• GET /users?skip=20&limit=10
  
  • skip: 건너뛸 row 수 (예: 0, 10, 20...)
  • limit: 한 번에 가져올 row 수

def get_users_paginated(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

---

• 페이징 예시 - 응답 구조

from pydantic import BaseModel

class PaginatedUsers(BaseModel):
    total: int
    skip: int
    limit: int
    data: List[UserRead]

@router.get("/users", response_model=PaginatedUsers)
def list_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    total = db.query(User).count()
    users = db.query(User).offset(skip).limit(limit).all()
    return PaginatedUsers(
        total=total,
        skip=skip,
        limit=limit,
        data=users
    )

• 응답 구조
  • {
      "total": 100,
      "skip": 20,
      "limit": 10,
      "data": [
        { "id": 21, "username": "hong", ... },
        ...
      ]
    }

---

• 조인 (클래스 정의)

# user.py
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String)
    posts = relationship("Post", back_populates="owner")

# post.py
class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="posts")

• 조인 사용 예시

# join(Post.owner) 은 내부적으로 posts.owner_id = users.id를 기준으로 조인
def get_posts_with_users(db: Session):
    return db.query(Post).join(Post.owner).all()
    
    
# 유저 정보 포함해서 가져오기 (query(Post, User) → 튜플로 결과 나옴)
def get_posts_with_user_info(db: Session):
    results = db.query(Post, User).join(User, Post.owner_id == User.id).all()

    for post, user in results:
        print(post.title, "-", user.username)
        
        
# outerjoin (LEFT JOIN) 예시
def get_all_users_with_or_without_posts(db: Session):
    return db.query(User).outerjoin(User.posts).all()
    
    
# Pydantic으로 응답 만들기
# joinedload() 써주면 N+1 문제 없이 User와 Post 같이 가져옴
class PostRead(BaseModel):
    id: int
    title: str

class UserReadWithPosts(BaseModel):
    id: int
    username: str
    posts: List[PostRead]

    class Config:
        orm_mode = True
        
@router.get("/users", response_model=List[UserReadWithPosts])
def list_users_with_posts(db: Session = Depends(get_db)):
    return db.query(User).options(joinedload(User.posts)).all()

---

2. SQLAlchemy + Raw SQL 방식 (가장 Low-level)

• SQL문을 직접 작성함
• text()로 감싸서 쿼리 실행
• 가장 직접적이지만, 유지보수가 어렵고 타입 지원도 없음

from sqlalchemy import text

def create_user_raw_sql(db, username: str, email: str):
    query = text("INSERT INTO users (username, email) VALUES (:username, :email)")
    db.execute(query, {"username": username, "email": email})
    db.commit()

---

3. SQLAlchemy Core 방식

• ORM 객체 없이 테이블을 직접 정의
• SQL을 Python 코드로 조립하는 방식
• 타입 지원이나 코드 자동화는 어느 정도 있음

from sqlalchemy import Table, Column, Integer, String, MetaData, insert

metadata = MetaData()

user_table = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("username", String),
    Column("email", String)
)

def create_user_core(db, username: str, email: str):
    stmt = insert(user_table).values(username=username, email=email)
    db.execute(stmt)
    db.commit()