설치 방법

pip install sqlalchemy

# PostgreSQL 연동 드라이버
pip install psycopg2-binary

# MySQL 연동 드라이버
pip install mysql-connector-python

핵심 구성 요소

• Engine: 데이터베이스 연결 관리자
• Session: 데이터베이스 작업 실행 컨텍스트
• Model: 테이블 구조를 정의하는 클래스
• Query: 데이터 조회를 위한 객체

데이터베이스 연결 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 연결 예시
db_engine = create_engine('sqlite:///app_data.db', echo=False)

# 세션 생성기 설정
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

# 세션 인스턴스화
db_session = SessionFactory()

데이터 모델 정의

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

ModelBase = declarative_base()

class Member(ModelBase):
    __tablename__ = 'members'
    
    member_id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(50), nullable=False)
    contact_email = Column(String(100), unique=True, index=True)
    
    articles = relationship("BlogPost", back_populates="writer")

class BlogPost(ModelBase):
    __tablename__ = 'blog_posts'
    
    post_id = Column(Integer, primary_key=True, index=True)
    post_title = Column(String(100), nullable=False)
    content_body = Column(String(500))
    writer_id = Column(Integer, ForeignKey('members.member_id'))
    
    writer = relationship("Member", back_populates="articles")
    categories = relationship("Category", secondary="post_categories", back_populates="posts")

class Category(ModelBase):
    __tablename__ = 'categories'
    
    cat_id = Column(Integer, primary_key=True, index=True)
    cat_name = Column(String(30), unique=True, nullable=False)
    
    posts = relationship("BlogPost", secondary="post_categories", back_populates="categories")

# 다대다 관계 매핑 테이블
class PostCategory(ModelBase):
    __tablename__ = 'post_categories'
    
    post_ref = Column(Integer, ForeignKey('blog_posts.post_id'), primary_key=True)
    cat_ref = Column(Integer, ForeignKey('categories.cat_id'), primary_key=True)

테이블 생성

ModelBase.metadata.create_all(bind=db_engine)

CRUD 작업

데이터 생성

# 단일 레코드 추가
new_member = Member(full_name="김철수", contact_email="chulsoo@example.kr")
db_session.add(new_member)
db_session.commit()

# 다중 레코드 일괄 추가
db_session.add_all([
    Member(full_name="이영희", contact_email="younghui@example.kr"),
    Member(full_name="박준수", contact_email="junsu@example.kr")
])
db_session.commit()

데이터 조회

# 전체 멤버 조회
all_members = db_session.query(Member).all()

# 조건별 필터링
filtered_members = db_session.query(Member).filter(Member.full_name.like("김%")).all()

데이터 수정

target_member = db_session.query(Member).get(1)
target_member.full_name = "김철수_수정"
db_session.commit()

데이터 삭제

del_member = db_session.query(Member).get(3)
db_session.delete(del_member)
db_session.commit()

고급 쿼리

from sqlalchemy import func

# 그룹화 집계
post_counts = db_session.query(
    Member.full_name, 
    func.count(BlogPost.post_id)
).join(BlogPost).group_by(Member.full_name).all()

# 다중 테이블 조인
joined_data = db_session.query(
    Member, BlogPost
).join(BlogPost).filter(BlogPost.post_title.contains("파이썬")).all()

관계형 데이터 처리

# 관계를 통한 객체 생성
author = Member(full_name="정민지", contact_email="minji@example.kr")
new_post = BlogPost(post_title="첫 게시글", content_body="안녕하세요!", writer=author)
db_session.add(new_post)
db_session.commit()

# 다대다 관계 설정
dev_category = Category(cat_name="개발")
db_category = Category(cat_name="데이터베이스")
new_post.categories.append(dev_category)
new_post.categories.append(db_category)
db_session.commit()

트랜잭션 관리

# 트랜잭션 컨텍스트 관리자
from contextlib import contextmanager

@contextmanager
def db_session_manager():
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()

# 사용 예시
with db_session_manager() as session:
    new_user = Member(full_name="트랜잭션테스트", contact_email="tx_test@example.kr")
    session.add(new_user)

성능 최적화 가이드

• 세션 수명 주기 관리: 요청 단위로 세션 생성/종료
• 즉시 로딩 활용: 관계형 데이터 조회 시 N+1 문제 방지
• 연결 풀 설정: 데이터베이스 연결 효율화
• 데이터 무결성: 애플리케이션 레벨 유효성 검증 구현