[FastAPI] SQLAlchemy 상세 - Join (6-6)

1. Join 기본 사용 방법

• join() 메서드를 이용해 직접 정의해 준다.
• 기본은 내부 조인(inner join) 이다.

with Session(engine) as session:
    # User와 Address 테이블을 조인하여 데이터를 조회
    stmt = select(User, Address).join(Address, User.id == Address.user_id)
    result = session.execute(stmt).all()

    for user, address in result:
        print(user.name, address.email_address)

 

• outerjoin() 메서드는 left join이다.

with Session(engine) as session:
    # User와 Address 테이블을 외부 조인하여 데이터를 조회
    stmt = select(User, Address).outerjoin(Address, User.id == Address.user_id)
    result = session.execute(stmt).all()

    for user, address in result:
        print(user.name, address.email_address if address else "No address")

 

• join 할 때, 값 선택

with Session(engine) as session:
    # User와 Address 테이블을 조인하여 특정 컬럼만 조회
    stmt = select(User.name, Address.email_address).join(Address, User.id == Address.user_id)
    result = session.execute(stmt).all()

    for name, email_address in result:
        print(name, email_address)

 

• 여러 테이블 join 해오기

with Session(engine) as session:
    # User, Address, Order 테이블을 조인하여 특정 컬럼만 조회
    stmt = (
        select(User.name, Address.email_address, Order.order_id)
        .join(Address, User.id == Address.user_id)
        .join(Order, User.id == Order.user_id)
    )
    result = session.execute(stmt).all()

    for name, email_address, order_id in result:
        print(name, email_address, order_id)

 

---

2. ORM을 이용한 기본 Join

먼저 두개의 테이블은

SQLAlchemy 모델에서 연결되어 있어야 한다.

relationship 에서 lazy옵션을 넣어 두면, 기본 join 옵션으로 이를 사용한다.

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    posts = relationship('Post', back_populates='user', lazy='select')  # Lazy Loading 설정

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship('User', back_populates='posts')

 

이제 단순히 User 를 불러오기만 해도

Join을 할 수 있다.

lazy옵션 에서 적용한 대로 자식 테이블을 조회한다.

stmt = select(User)
result = session.execute(stmt)

# User를 로드한 후 Post에 접근할 때 별도의 쿼리가 실행됨
for user in result.scalars():
    print(user.name)  # User 쿼리만 실행
    print(user.posts)  # 여기서 Post를 가져오기 위한 SELECT 쿼리가 실행됨

 

---

3. ORM 방식 Join 옵션 적용

lazy 설정을 무시하고, 명시적으로 관계 데이터를 한 번의 쿼리로 가져오고 싶다면

.options() 를 사용해 준다.

 

joinedload() 를 사용해주면

관계 데이터를 한번에 가져온다. Eager Loading 방식

from sqlalchemy.orm import joinedload

# 사용자와 함께 관련 게시물 로드
stmt = select(User).options(joinedload(User.posts))

result = session.execute(stmt)

# 데이터 출력
for user in result.scalars():  # .scalars()로 User 객체만 가져옴
    print(f"User: {user.name}")
    for post in user.posts:
        print(f"  Post: {post.title}")

 

selectinload() 

기본 객체 쿼리 후 IN 쿼리로 추가 데이터를 가져오는 Eager Loading 방식.

• 특징: joinedload()와 유사하지만, 조인을 사용하지 않아서 더 효율적일 수 있다. 조인이 복잡하거나 너무 많은 데이터를 가져올 위험이 있을 때 사용한다.

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.posts))
result = session.execute(stmt)

# User와 관련된 Post를 별도의 IN 쿼리로 가져옴
for user in result.scalars():
    print(user.name, [post.title for post in user.posts])

 

subqueryload()

관련 데이터를 서브쿼리를 사용해 가져오는 방식

조인이 비효율적일 때 사용

from sqlalchemy.orm import subqueryload

stmt = select(User).options(subqueryload(User.posts))
result = session.execute(stmt)

# 서브쿼리로 User와 관련된 Post를 가져옴
for user in result.scalars():
    print(user.name, [post.title for post in user.posts])

 

lazyload()

관계 데이터를 필요할 때 지연 로드. 기본적으로 SQLAlchemy에서 사용하는 방식.

from sqlalchemy.orm import lazyload

stmt = select(User).options(lazyload(User.posts))
result = session.execute(stmt)

# User만 로드하고, Post는 접근할 때 로드됨
for user in result.scalars():
    print(user.name)  # Post는 아직 로드되지 않음
    print([post.title for post in user.posts])  # 여기서 Post가 로드됨

 

defer()

특정 컬럼의 로드를 지연시킨다. 필요할 때만 데이터를 가져옴

크기가 큰 텍스트나 바이너리 데이터를 기본적으로 로드하지 않으려는 경우 사용

from sqlalchemy.orm import defer

stmt = select(User).options(defer(User.name))
result = session.execute(stmt)

# User의 'name' 컬럼은 접근할 때만 로드됨
for user in result.scalars():
    print(user.id)  # 로드됨
    print(user.name)  # 여기서 name이 로드됨

 

undefer()

deferred로 설정된 컬럼을 명시적으로 로드

필요에 따라 지연 로드된 컬럼을 미리 로드하려는 경우.

from sqlalchemy.orm import undefer

stmt = select(User).options(undefer(User.name))
result = session.execute(stmt)

# User의 'name' 컬럼이 즉시 로드됨
for user in result.scalars():
    print(user.name)

 

raiseload()

관계 데이터를 접근하려고 하면 예외를 발생

예상치 못한 지연 로드를 방지하기 위한 디버깅 도구로 사용

from sqlalchemy.orm import raiseload

stmt = select(User).options(raiseload(User.posts))
result = session.execute(stmt)

# User는 로드되지만, Post에 접근하면 예외가 발생
for user in result.scalars():
    print(user.name)
    print(user.posts)  # 여기에 접근 시 예외 발생

 

load_only()

특정 컬럼만 로드. 나머지 컬럼은 로드되지 않음

from sqlalchemy.orm import load_only

stmt = select(User).options(load_only(User.name))
result = session.execute(stmt)

# User의 'name' 컬럼만 로드됨
for user in result.scalars():
    print(user.name)
    # print(user.id)  # 접근 시 예외 발생

 

contains_eager()

미리 로드한 데이터를 활용하여 Eager Loading을 수행

복잡한 쿼리에서 이미 로드된 데이터를 활용해 추가 쿼리를 방지하고 싶을 때

stmt = (
    select(User)
    .join(User.posts)
    .options(contains_eager(User.posts))
)
result = session.execute(stmt)

# User와 Post를 이미 로드된 데이터로 연결
for user in result.scalars():
    print(user.name, [post.title for post in user.posts])

 

with_loader_criteria()

관계 데이터를 로드할 때 필터링 조건을 추가로 지정

from sqlalchemy.orm import with_loader_criteria

stmt = select(User).options(
    with_loader_criteria(Post, Post.title.like('%example%'))
)
result = session.execute(stmt)

# 조건에 맞는 Post만 로드됨
for user in result.scalars():
    print(user.name, [post.title for post in user.posts])

 

 

요약

• Eager Loading: joinedload, selectinload, subqueryload

• Lazy Loading: lazyload, defer, undefer, raiseload

• 컬럼 제어: load_only, defer, undefer

• 디버깅: raiseload

• 필터링 : with_loader_criteria

• 특수 상황: contains_eager

 

---

4. orm.join (명시적 ORM Join)

SQLAlchemy ORM의 join() 메서드 대신 orm.join()을 사용하여 ORM 관계를 기반으로 조인을 수행

from sqlalchemy.orm import join

stmt = select(User).select_from(join(User, Post, User.id == Post.user_id))
result = session.execute(stmt)

for user in result.scalars():
    print(user.name)

 

---

5. 다중 관계 조인

여러 관계를 한번에 조인할 수 있다.

stmt = (
    select(User.name, Post.title, Comment.text)
    .join(Post, User.id == Post.user_id)
    .join(Comment, Post.id == Comment.post_id)
)
result = session.execute(stmt)

for row in result:
    print(row.name, row.title, row.text)

 

---

6. 서브쿼리 생성과 조인

SQLAlchemy에서 서브쿼리를 생성하려면 select()로 쿼리를 작성한다.

그 다음으로 .subquery()를 호출한다.

그런 다음, 이 서브쿼리를 다른 테이블과 join()을 통해 결합한다.

from sqlalchemy import select, func

# 서브쿼리 생성
post_count_subquery = (
    select(
        Post.user_id,
        func.count(Post.id).label("post_count")
    )
    .group_by(Post.user_id)
    .subquery()
)

# 서브쿼리를 User와 조인
stmt = (
    select(User.name, post_count_subquery.c.post_count)
    .join(post_count_subquery, User.id == post_count_subquery.c.user_id)
)

result = session.execute(stmt)

# 결과 출력
for row in result:
    print(row.name, row.post_count)

 

---

7. 같은 테이블 조인

같은 테이블을 여러 번 조인해야 할 경우 별칭(alias)을 만들어 관리

from sqlalchemy.orm import aliased

PostAlias = aliased(Post)

stmt = (
    select(User.name, Post.title, PostAlias.title.label('other_post'))
    .join(Post, User.id == Post.user_id)
    .join(PostAlias, User.id == PostAlias.user_id)
)
result = session.execute(stmt)

for row in result:
    print(row.name, row.title, row.other_post)