pip install sqlalchemy==2.3.0

2025년 3월 기준 최신 버전인 2.3.0을 설치하는 명령어야.

# 가상 환경 생성
python -m venv sqlalchemy-env

# 가상 환경 활성화 (Windows)
sqlalchemy-env\Scripts\activate

# 가상 환경 활성화 (macOS/Linux)
source sqlalchemy-env/bin/activate

# SQLAlchemy 설치
pip install sqlalchemy==2.3.0

from sqlalchemy import create_engine, text

# SQLite 메모리 데이터베이스 연결
engine = create_engine('sqlite:///:memory:')

# 연결 테스트
with engine.connect() as conn:
    result = conn.execute(text("SELECT 'Hello, SQLAlchemy!'"))
    print(result.all())

# 출력: [('Hello, SQLAlchemy!',)]

from sqlalchemy import create_engine

# SQLite 데이터베이스 엔진 생성
engine = create_engine('sqlite:///example.db')

# PostgreSQL 데이터베이스 엔진 생성
# engine = create_engine('postgresql://username:password@localhost:5432/dbname')

# MySQL 데이터베이스 엔진 생성
# engine = create_engine('mysql+pymysql://username:password@localhost:3306/dbname')

from sqlalchemy import text

# 연결 얻기
with engine.connect() as conn:
    # 간단한 쿼리 실행
    result = conn.execute(text("SELECT * FROM users"))
    
    # 결과 처리
    for row in result:
        print(row)

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

# 메타데이터 객체 생성
metadata = MetaData()

# 테이블 정의
users = Table(
    'users', 
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('email', String(100))
)

# 메타데이터를 통해 데이터베이스에 테이블 생성
metadata.create_all(engine)

from sqlalchemy.orm import Session

# 세션 생성
with Session(engine) as session:
    # 트랜잭션 시작
    
    # 데이터 조회
    users = session.query(User).all()
    
    # 새 사용자 추가
    new_user = User(name="홍길동", email="hong@example.com")
    session.add(new_user)
    
    # 변경사항 커밋
    session.commit()

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, ForeignKey
from typing import List, Optional
from datetime import datetime

# 베이스 클래스 정의
class Base(DeclarativeBase):
    pass

# 사용자 모델
class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    # Optional 필드 (NULL 허용)
    full_name: Mapped[Optional[str]] = mapped_column(String(100))
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

# 게시글 모델
class Post(Base):
    __tablename__ = 'posts'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String(2000))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    
    def __repr__(self):
        return f"<Post(title='{self.title}', user_id={self.user_id})>"

from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    full_name: Mapped[Optional[str]] = mapped_column(String(100))
    
    # 관계 설정: 한 사용자는 여러 게시글을 가질 수 있음
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="author")
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

class Post(Base):
    __tablename__ = 'posts'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String(2000))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    
    # 관계 설정: 각 게시글은 한 명의 작성자를 가짐
    author: Mapped["User"] = relationship("User", back_populates="posts")
    
    def __repr__(self):
        return f"<Post(title='{self.title}', user_id={self.user_id})>"

# 사용자의 모든 게시글 접근
user = session.query(User).first()
for post in user.posts:
    print(post.title)

# 게시글의 작성자 접근
post = session.query(Post).first()
print(post.author.username)

# 태그와 게시글 간의 다대다 관계
# 중간 테이블 정의
post_tags = Table(
    'post_tags',
    Base.metadata,
    Column('post_id', ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', ForeignKey('tags.id'), primary_key=True)
)

class Tag(Base):
    __tablename__ = 'tags'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    
    # 다대다 관계 설정
    posts: Mapped[List["Post"]] = relationship("Post", secondary=post_tags, back_populates="tags")
    
    def __repr__(self):
        return f"<Tag(name='{self.name}')>"

class Post(Base):
    # 기존 코드 생략...
    
    # 다대다 관계 추가
    tags: Mapped[List["Tag"]] = relationship("Tag", secondary=post_tags, back_populates="posts")

from sqlalchemy import create_engine

# 엔진 생성
engine = create_engine('sqlite:///blog.db')

# 데이터베이스에 테이블 생성
Base.metadata.create_all(engine)

from sqlalchemy.orm import Session

# 세션 생성
with Session(engine) as session:
    # 새 사용자 생성
    user = User(username="python_lover", email="python@example.com", full_name="파이썬 러버")
    
    # 게시글 생성
    post1 = Post(title="SQLAlchemy 시작하기", content="SQLAlchemy는 정말 강력한 ORM입니다...")
    post2 = Post(title="파이썬 비동기 프로그래밍", content="asyncio를 사용하면...")
    
    # 사용자와 게시글 연결
    user.posts.append(post1)
    user.posts.append(post2)
    
    # 세션에 객체 추가
    session.add(user)
    
    # 변경사항 커밋
    session.commit()

from sqlalchemy.orm import Session

with Session(engine) as session:
    # 새 태그 생성
    tag1 = Tag(name="Python")
    tag2 = Tag(name="SQLAlchemy")
    tag3 = Tag(name="Database")
    
    # 새 사용자 생성
    user = User(
        username="data_scientist",
        email="data@example.com",
        full_name="데이터 과학자"
    )
    
    # 새 게시글 생성
    post = Post(
        title="데이터 분석과 SQLAlchemy",
        content="데이터 분석 프로젝트에서 SQLAlchemy를 활용하는 방법...",
        author=user  # 관계를 통한 할당
    )
    
    # 게시글에 태그 추가
    post.tags = [tag1, tag2, tag3]
    
    # 세션에 객체 추가
    session.add_all([tag1, tag2, tag3, user, post])
    
    # 변경사항 커밋
    session.commit()
    
    # 생성된 ID 확인
    print(f"생성된 사용자 ID: {user.id}")
    print(f"생성된 게시글 ID: {post.id}")

from sqlalchemy import select

with Session(engine) as session:
    # 모든 사용자 조회
    stmt = select(User)
    result = session.execute(stmt)
    users = result.scalars().all()
    
    for user in users:
        print(f"사용자: {user.username}, 이메일: {user.email}")
    
    # 특정 조건으로 사용자 조회
    stmt = select(User).where(User.username == "python_lover")
    result = session.execute(stmt)
    user = result.scalar_one_or_none()  # 결과가 없으면 None 반환
    
    if user:
        print(f"찾은 사용자: {user.username}")
        
        # 관계를 통한 게시글 접근
        print(f"{user.username}의 게시글:")
        for post in user.posts:
            print(f"- {post.title}")

from sqlalchemy import select, func

with Session(engine) as session:
    # 게시글 수가 많은 순으로 사용자 정렬
    stmt = select(User, func.count(Post.id).label("post_count")) \
        .join(User.posts) \
        .group_by(User.id) \
        .order_by(func.count(Post.id).desc())
    
    result = session.execute(stmt)
    
    for user, post_count in result:
        print(f"사용자: {user.username}, 게시글 수: {post_count}")
    
    # 특정 태그가 있는 게시글 찾기
    stmt = select(Post) \
        .join(Post.tags) \
        .where(Tag.name == "SQLAlchemy")
    
    result = session.execute(stmt)
    posts = result.scalars().all()
    
    print("SQLAlchemy 태그가 있는 게시글:")
    for post in posts:
        print(f"- {post.title} (작성자: {post.author.username})")

from sqlalchemy import update

with Session(engine) as session:
    # 방법 1: 객체 속성 변경
    stmt = select(User).where(User.username == "python_lover")
    result = session.execute(stmt)
    user = result.scalar_one()
    
    # 속성 변경
    user.email = "new_email@example.com"
    user.full_name = "파이썬 마스터"
    
    # 변경사항 커밋
    session.commit()
    
    # 방법 2: UPDATE 문 실행
    stmt = update(Post) \
        .where(Post.user_id == user.id) \
        .values(content=Post.content + "\n\n업데이트: 내용이 수정되었습니다.")
    
    session.execute(stmt)
    session.commit()

from sqlalchemy import delete

with Session(engine) as session:
    # 방법 1: 객체 삭제
    stmt = select(Post).where(Post.title == "삭제할 게시글")
    result = session.execute(stmt)
    post = result.scalar_one_or_none()
    
    if post:
        session.delete(post)
        session.commit()
        print("게시글이 삭제되었습니다.")
    
    # 방법 2: DELETE 문 실행
    stmt = delete(Post).where(Post.created_at < datetime(2024, 1, 1))
    result = session.execute(stmt)
    session.commit()
    print(f"{result.rowcount}개의 오래된 게시글이 삭제되었습니다.")

try:
    with Session(engine) as session:
        # 여러 작업 수행
        user = User(username="new_user", email="new@example.com")
        session.add(user)
        
        post = Post(title="새 게시글", content="내용...", author=user)
        session.add(post)
        
        # 의도적으로 오류 발생 (예시)
        if post.title == "새 게시글":
            raise ValueError("테스트 오류")
        
        # 이 부분은 오류 때문에 실행되지 않음
        session.commit()
except ValueError as e:
    print(f"오류 발생: {e}")
    print("트랜잭션이 롤백되었습니다.")

# 롤백 확인
with Session(engine) as session:
    stmt = select(User).where(User.username == "new_user")
    result = session.execute(stmt)
    user = result.scalar_one_or_none()
    print(f"사용자 존재 여부: {user is not None}")  # False 출력

with Session(engine) as session:
    try:
        # 트랜잭션 시작
        
        # 작업 수행
        user = User(username="transaction_test", email="test@example.com")
        session.add(user)
        
        # 중간 저장점 생성
        savepoint = session.begin_nested()
        
        try:
            # 위험한 작업
            post = Post(title="", content="")  # 제목이 비어 있어 오류 발생 가능
            session.add(post)
            
            # 이 부분은 오류 때문에 실행되지 않을 수 있음
            savepoint.commit()
        except Exception as e:
            # 저장점까지만 롤백
            savepoint.rollback()
            print(f"게시글 추가 중 오류 발생: {e}")
        
        # 사용자는 여전히 추가됨
        session.commit()
    except Exception as e:
        # 전체 트랜잭션 롤백
        session.rollback()
        print(f"전체 트랜잭션 오류: {e}")

from sqlalchemy import select, func

with Session(engine) as session:
    # 내부 조인(INNER JOIN): 사용자와 그들의 게시글 조회
    stmt = select(User.username, Post.title, Post.created_at) \
        .join(Post, User.id == Post.user_id) \
        .order_by(Post.created_at.desc())
    
    result = session.execute(stmt)
    
    print("사용자와 게시글:")
    for username, title, created_at in result:
        print(f"{username}: {title} ({created_at})")
    
    # 외부 조인(LEFT OUTER JOIN): 게시글이 없는 사용자도 포함
    stmt = select(User.username, func.count(Post.id).label("post_count")) \
        .outerjoin(Post) \
        .group_by(User.username) \
        .order_by(func.count(Post.id).desc())
    
    result = session.execute(stmt)
    
    print("\n사용자별 게시글 수:")
    for username, post_count in result:
        print(f"{username}: {post_count}개")

from sqlalchemy import select, func, and_, or_, desc

with Session(engine) as session:
    # 서브쿼리: 각 사용자의 최신 게시글 찾기
    subq = select(Post.user_id, func.max(Post.created_at).label("latest_post")) \
        .group_by(Post.user_id) \
        .subquery()
    
    # 메인 쿼리: 각 사용자의 최신 게시글 정보 조회
    stmt = select(User.username, Post.title, Post.created_at) \
        .join(subq, and_(
            User.id == subq.c.user_id,
            Post.created_at == subq.c.latest_post
        )) \
        .join(Post, User.id == Post.user_id)
    
    result = session.execute(stmt)
    
    print("각 사용자의 최신 게시글:")
    for username, title, created_at in result:
        print(f"{username}: {title} ({created_at})")

from sqlalchemy import select, and_, or_, not_, desc, asc

with Session(engine) as session:
    # 복잡한 조건으로 필터링
    stmt = select(Post) \
        .where(
            and_(
                Post.created_at >= datetime(2024, 1, 1),
                or_(
                    Post.title.like('%Python%'),
                    Post.title.like('%SQLAlchemy%')
                ),
                not_(Post.content == '')
            )
        ) \
        .order_by(desc(Post.created_at))
    
    result = session.execute(stmt)
    posts = result.scalars().all()
    
    print("2024년 이후 Python 또는 SQLAlchemy 관련 게시글:")
    for post in posts:
        print(f"- {post.title} ({post.created_at})")
    
    # IN 연산자 사용
    popular_tags = ["Python", "Database", "ORM"]
    stmt = select(Post) \
        .join(Post.tags) \
        .where(Tag.name.in_(popular_tags)) \
        .order_by(Post.created_at.desc())
    
    result = session.execute(stmt)
    posts = result.scalars().all()
    
    print("\n인기 태그가 있는 게시글:")
    for post in posts:
        tag_names = [tag.name for tag in post.tags]
        print(f"- {post.title} (태그: {', '.join(tag_names)})")

from sqlalchemy import select, func, extract

with Session(engine) as session:
    # 월별 게시글 수 집계
    stmt = select(
            extract('year', Post.created_at).label('year'),
            extract('month', Post.created_at).label('month'),
            func.count(Post.id).label('post_count')
        ) \
        .group_by('year', 'month') \
        .order_by('year', 'month')
    
    result = session.execute(stmt)
    
    print("월별 게시글 수:")
    for year, month, count in result:
        print(f"{int(year)}년 {int(month)}월: {count}개")
    
    # 태그별 게시글 수와 평균 길이
    stmt = select(
            Tag.name,
            func.count(Post.id).label('post_count'),
            func.avg(func.length(Post.content)).label('avg_length')
        ) \
        .join(Post.tags) \
        .group_by(Tag.name) \
        .having(func.count(Post.id) > 1)  # 2개 이상의 게시글이 있는 태그만
        .order_by(func.count(Post.id).desc())
    
    result = session.execute(stmt)
    
    print("\n태그별 통계:")
    for tag_name, post_count, avg_length in result:
        print(f"{tag_name}: {post_count}개 게시글, 평균 길이: {int(avg_length)}자")

def get_posts_page(session, page=1, per_page=10):
    # 페이지 번호는 1부터 시작
    offset_val = (page - 1) * per_page
    
    # 게시글 조회 쿼리
    stmt = select(Post) \
        .order_by(Post.created_at.desc()) \
        .offset(offset_val) \
        .limit(per_page)
    
    result = session.execute(stmt)
    posts = result.scalars().all()
    
    # 전체 게시글 수 조회
    count_stmt = select(func.count()).select_from(Post)
    total = session.execute(count_stmt).scalar()
    
    # 전체 페이지 수 계산
    total_pages = (total + per_page - 1) // per_page
    
    return {
        'posts': posts,
        'page': page,
        'per_page': per_page,
        'total': total,
        'total_pages': total_pages,
        'has_next': page < total_pages,
        'has_prev': page > 1
    }

# 사용 예시
with Session(engine) as session:
    # 2페이지, 페이지당 5개 항목
    page_data = get_posts_page(session, page=2, per_page=5)
    
    print(f"페이지 {page_data['page']}/{page_data['total_pages']} (총 {page_data['total']}개)")
    
    for post in page_data['posts']:
        print(f"- {post.title} ({post.created_at})")
    
    if page_data['has_prev']:
        print("이전 페이지 있음")
    
    if page_data['has_next']:
        print("다음 페이지 있음")

from sqlalchemy import select
from sqlalchemy.orm import selectinload, joinedload

with Session(engine) as session:
    # 지연 로딩 (기본값) - N+1 쿼리 문제 발생 가능
    stmt = select(User).where(User.username.like('%user%'))
    result = session.execute(stmt)
    users = result.scalars().all()
    
    # 각 사용자의 게시글에 접근할 때마다 추가 쿼리 실행
    for user in users:
        print(f"{user.username}의 게시글: {len(user.posts)}개")
    
    # 즉시 로딩 (selectinload) - 컬렉션에 적합
    stmt = select(User) \
        .where(User.username.like('%user%')) \
        .options(selectinload(User.posts))
    
    result = session.execute(stmt)
    users = result.scalars().all()
    
    # 이미 로드된 게시글에 접근 (추가 쿼리 없음)
    for user in users:
        print(f"{user.username}의 게시글: {len(user.posts)}개")
    
    # 즉시 로딩 (joinedload) - 단일 관계에 적합
    stmt = select(Post) \
        .options(joinedload(Post.author)) \
        .where(Post.title.like('%Python%'))
    
    result = session.execute(stmt)
    posts = result.scalars().all()
    
    # 이미 로드된 작성자에 접근 (추가 쿼리 없음)
    for post in posts:
        print(f"게시글: {post.title}, 작성자: {post.author.username}")

from sqlalchemy import select, func, CTE

with Session(engine) as session:
    # 사용자별 게시글 수를 계산하는 CTE
    user_post_count = select(
            User.id.label('user_id'),
            User.username,
            func.count(Post.id).label('post_count')
        ) \
        .join(Post, User.id == Post.user_id) \
        .group_by(User.id, User.username) \
        .cte('user_post_count')
    
    # 평균 이상의 게시글을 작성한 사용자 찾기
    avg_post_count = select(func.avg(user_post_count.c.post_count)).scalar_subquery()
    
    stmt = select(
            user_post_count.c.username,
            user_post_count.c.post_count
        ) \
        .where(user_post_count.c.post_count > avg_post_count) \
        .order_by(user_post_count.c.post_count.desc())
    
    result = session.execute(stmt)
    
    print("평균 이상의 게시글을 작성한 사용자:")
    for username, post_count in result:
        print(f"{username}: {post_count}개")

blog_project/
├── models.py      # 데이터베이스 모델
├── database.py    # 데이터베이스 연결 및 세션 관리
├── repositories.py # 데이터 액세스 레이어
├── services.py    # 비즈니스 로직
└── main.py        # 애플리케이션 진입점

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

# 데이터베이스 URL (환경에 맞게 수정)
DATABASE_URL = "sqlite:///blog.db"

# 엔진 생성
engine = create_engine(DATABASE_URL, echo=True)

# 세션 팩토리 생성
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    """
    데이터베이스 세션을 생성하고 관리하는 함수
    컨텍스트 매니저로 사용 가능
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_db_session():
    """
    일반 함수에서 세션을 얻기 위한 헬퍼 함수
    """
    return next(get_db())

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, Text, ForeignKey, Table, Column
from typing import List, Optional
from datetime import datetime

class Base(DeclarativeBase):
    pass

# 태그와 게시글 간의 다대다 관계를 위한 연결 테이블
post_tags = Table(
    'post_tags',
    Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    password_hash: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    is_active: Mapped[bool] = mapped_column(default=True)
    
    # 관계 설정
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="author", cascade="all, delete-orphan")
    comments: Mapped[List["Comment"]] = relationship("Comment", back_populates="author", cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

class Post(Base):
    __tablename__ = 'posts'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    content: Mapped[str] = mapped_column(Text)
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    updated_at: Mapped[Optional[datetime]] = mapped_column(nullable=True)
    is_published: Mapped[bool] = mapped_column(default=True)
    view_count: Mapped[int] = mapped_column(default=0)
    
    # 관계 설정
    author: Mapped["User"] = relationship("User", back_populates="posts")
    comments: Mapped[List["Comment"]] = relationship("Comment", back_populates="post", cascade="all, delete-orphan")
    tags: Mapped[List["Tag"]] = relationship("Tag", secondary=post_tags, back_populates="posts")
    
    def __repr__(self):
        return f"<Post(title='{self.title}', author='{self.author.username}')>"

class Comment(Base):
    __tablename__ = 'comments'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str] = mapped_column(Text)
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    post_id: Mapped[int] = mapped_column(ForeignKey('posts.id'))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    
    # 관계 설정
    author: Mapped["User"] = relationship("User", back_populates="comments")
    post: Mapped["Post"] = relationship("Post", back_populates="comments")
    
    def __repr__(self):
        return f"<Comment(author='{self.author.username}', post_id={self.post_id})>"

class Tag(Base):
    __tablename__ = 'tags'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    
    # 관계 설정
    posts: Mapped[List["Post"]] = relationship("Post", secondary=post_tags, back_populates="tags")
    
    def __repr__(self):
        return f"<Tag(name='{self.name}')>"

from sqlalchemy import select, func, desc, update, delete
from sqlalchemy.orm import Session, selectinload
from typing import List, Optional, Dict, Any
from datetime import datetime

from models import User, Post, Comment, Tag

class UserRepository:
    def __init__(self, session: Session):
        self.session = session
    
    def create(self, username: str, email: str, password_hash: str) -> User:
        user = User(
            username=username,
            email=email,
            password_hash=password_hash
        )
        self.session.add(user)
        self.session.commit()
        return user
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        stmt = select(User).where(User.id == user_id)
        result = self.session.execute(stmt)
        return result.scalar_one_or_none()
    
    def get_by_username(self, username: str) -> Optional[User]:
        stmt = select(User).where(User.username == username)
        result = self.session.execute(stmt)
        return result.scalar_one_or_none()
    
    def update(self, user_id: int, data: Dict[str, Any]) -> Optional[User]:
        user = self.get_by_id(user_id)
        if not user:
            return None
        
        for key, value in data.items():
            if hasattr(user, key):
                setattr(user, key, value)
        
        self.session.commit()
        return user
    
    def delete(self, user_id: int) -> bool:
        user = self.get_by_id(user_id)
        if not user:
            return False
        
        self.session.delete(user)
        self.session.commit()
        return True

class PostRepository:
    def __init__(self, session: Session):
        self.session = session
    
    def create(self, title: str, content: str, user_id: int, tags: List[str] = None) -> Post:
        # 게시글 생성
        post = Post(
            title=title,
            content=content,
            user_id=user_id
        )
        
        # 태그 처리
        if tags:
            for tag_name in tags:
                # 기존 태그 찾기 또는 새로 생성
                stmt = select(Tag).where(Tag.name == tag_name)
                result = self.session.execute(stmt)
                tag = result.scalar_one_or_none()
                
                if not tag:
                    tag = Tag(name=tag_name)
                    self.session.add(tag)
                
                post.tags.append(tag)
        
        self.session.add(post)
        self.session.commit()
        return post
    
    def get_by_id(self, post_id: int, with_author: bool = False, with_tags: bool = False) -> Optional[Post]:
        stmt = select(Post).where(Post.id == post_id)
        
        if with_author:
            stmt = stmt.options(selectinload(Post.author))
        
        if with_tags:
            stmt = stmt.options(selectinload(Post.tags))
        
        result = self.session.execute(stmt)
        post = result.scalar_one_or_none()
        
        if post:
            # 조회수 증가
            post.view_count += 1
            self.session.commit()
        
        return post
    
    def get_all(self, page: int = 1, per_page: int = 10, with_author: bool = False, with_tags: bool = False) -> Dict[str, Any]:
        # 오프셋 계산
        offset_val = (page - 1) * per_page
        
        # 기본 쿼리
        stmt = select(Post).where(Post.is_published == True).order_by(desc(Post.created_at))
        
        # 관계 로딩 옵션 추가
        if with_author:
            stmt = stmt.options(selectinload(Post.author))
        
        if with_tags:
            stmt = stmt.options(selectinload(Post.tags))
        
        # 페이지네이션 적용
        stmt = stmt.offset(offset_val).limit(per_page)
        
        # 쿼리 실행
        result = self.session.execute(stmt)
        posts = result.scalars().all()
        
        # 전체 게시글 수 조회
        count_stmt = select(func.count()).select_from(Post).where(Post.is_published == True)
        total = self.session.execute(count_stmt).scalar()
        
        # 전체 페이지 수 계산
        total_pages = (total + per_page - 1) // per_page
        
        return {
            'posts': posts,
            'page': page,
            'per_page': per_page,
            'total': total,
            'total_pages': total_pages,
            'has_next': page < total_pages,
            'has_prev': page > 1
        }
    
    def get_by_tag(self, tag_name: str, page: int = 1, per_page: int = 10) -> Dict[str, Any]:
        # 오프셋 계산
        offset_val = (page - 1) * per_page
        
        # 태그로 게시글 조회
        stmt = select(Post) \
            .join(Post.tags) \
            .where(Tag.name == tag_name) \
            .where(Post.is_published == True) \
            .order_by(desc(Post.created_at)) \
            .offset(offset_val) \
            .limit(per_page)
        
        result = self.session.execute(stmt)
        posts = result.scalars().all()
        
        # 전체 게시글 수 조회
        count_stmt = select(func.count()) \
            .select_from(Post) \
            .join(Post.tags) \
            .where(Tag.name == tag_name) \
            .where(Post.is_published == True)
        
        total = self.session.execute(count_stmt).scalar()
        
        # 전체 페이지 수 계산
        total_pages = (total + per_page - 1) // per_page
        
        return {
            'posts': posts,
            'page': page,
            'per_page': per_page,
            'total': total,
            'total_pages': total_pages,
            'has_next': page < total_pages,
            'has_prev': page > 1,
            'tag_name': tag_name
        }
    
    def update(self, post_id: int, data: Dict[str, Any]) -> Optional[Post]:
        post = self.get_by_id(post_id)
        if not post:
            return None
        
        # 업데이트 시간 설정
        data['updated_at'] = datetime.now()
        
        # 태그 처리
        tags = data.pop('tags', None)
        
        # 기본 필드 업데이트
        for key, value in data.items():
            if hasattr(post, key):
                setattr(post, key, value)
        
        # 태그 업데이트
        if tags is not None:
            # 기존 태그 모두 제거
            post.tags.clear()
            
            # 새 태그 추가
            for tag_name in tags:
                stmt = select(Tag).where(Tag.name == tag_name)
                result = self.session.execute(stmt)
                tag = result.scalar_one_or_none()
                
                if not tag:
                    tag = Tag(name=tag_name)
                    self.session.add(tag)
                
                post.tags.append(tag)
        
        self.session.commit()
        return post
    
    def delete(self, post_id: int) -> bool:
        post = self.get_by_id(post_id)
        if not post:
            return False
        
        self.session.delete(post)
        self.session.commit()
        return True

class CommentRepository:
    def __init__(self, session: Session):
        self.session = session
    
    def create(self, content: str, user_id: int, post_id: int) -> Comment:
        comment = Comment(
            content=content,
            user_id=user_id,
            post_id=post_id
        )
        self.session.add(comment)
        self.session.commit()
        return comment
    
    def get_by_post(self, post_id: int) -> List[Comment]:
        stmt = select(Comment) \
            .where(Comment.post_id == post_id) \
            .options(selectinload(Comment.author)) \
            .order_by(Comment.created_at)
        
        result = self.session.execute(stmt)
        return result.scalars().all()
    
    def delete(self, comment_id: int) -> bool:
        stmt = select(Comment).where(Comment.id == comment_id)
        result = self.session.execute(stmt)
        comment = result.scalar_one_or_none()
        
        if not comment:
            return False
        
        self.session.delete(comment)
        self.session.commit()
        return True

class TagRepository:
    def __init__(self, session: Session):
        self.session = session
    
    def get_all(self) -> List[Tag]:
        stmt = select(Tag).order_by(Tag.name)
        result = self.session.execute(stmt)
        return result.scalars().all()
    
    def get_popular_tags(self, limit: int = 10) -> List[Dict[str, Any]]:
        # 게시글 수가 많은 순으로 태그 조회
        stmt = select(
                Tag.name,
                func.count(Post.id).label('post_count')
            ) \
            .join(Tag.posts) \
            .group_by(Tag.name) \
            .order_by(desc(func.count(Post.id))) \
            .limit(limit)
        
        result = self.session.execute(stmt)
        
        return [{'name': name, 'post_count': count} for name, count in result]

from sqlalchemy.orm import Session
from typing import List, Dict, Any, Optional
import hashlib
import os

from repositories import UserRepository, PostRepository, CommentRepository, TagRepository

class UserService:
    def __init__(self, session: Session):
        self.repo = UserRepository(session)
    
    def register(self, username: str, email: str, password: str) -> Dict[str, Any]:
        # 사용자 이름 중복 확인
        existing_user = self.repo.get_by_username(username)
        if existing_user:
            return {'success': False, 'message': '이미 사용 중인 사용자 이름입니다.'}
        
        # 비밀번호 해싱
        password_hash = self._hash_password(password)
        
        # 사용자 생성
        user = self.repo.create(username, email, password_hash)
        
        return {
            'success': True,
            'user': {
                'id': user.id,
                'username': user.username,
                'email': user.email
            }
        }
    
    def authenticate(self, username: str, password: str) -> Dict[str, Any]:
        user = self.repo.get_by_username(username)
        if not user:
            return {'success': False, 'message': '사용자를 찾을 수 없습니다.'}
        
        # 비밀번호 확인
        if user.password_hash != self._hash_password(password):
            return {'success': False, 'message': '비밀번호가 일치하지 않습니다.'}
        
        return {
            'success': True,
            'user': {
                'id': user.id,
                'username': user.username,
                'email': user.email
            }
        }
    
    def get_profile(self, user_id: int) -> Dict[str, Any]:
        user = self.repo.get_by_id(user_id)
        if not user:
            return {'success': False, 'message': '사용자를 찾을 수 없습니다.'}
        
        return {
            'success': True,
            'user': {
                'id': user.id,
                'username': user.username,
                'email': user.email,
                'created_at': user.created_at,
                'post_count': len(user.posts),
                'comment_count': len(user.comments)
            }
        }
    
    def _hash_password(self, password: str) -> str:
        # 실제 애플리케이션에서는 더 안전한 방식(bcrypt 등) 사용 권장
        salt = os.environ.get('PASSWORD_SALT', 'default_salt')
        return hashlib.sha256((password + salt).encode()).hexdigest()

class BlogService:
    def __init__(self, session: Session):
        self.post_repo = PostRepository(session)
        self.comment_repo = CommentRepository(session)
        self.tag_repo = TagRepository(session)
    
    def create_post(self, title: str, content: str, user_id: int, tags: List[str] = None) -> Dict[str, Any]:
        try:
            post = self.post_repo.create(title, content, user_id, tags)
            return {
                'success': True,
                'post': {
                    'id': post.id,
                    'title': post.title,
                    'created_at': post.created_at
                }
            }
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def get_post(self, post_id: int) -> Dict[str, Any]:
        post = self.post_repo.get_by_id(post_id, with_author=True, with_tags=True)
        if not post:
            return {'success': False, 'message': '게시글을 찾을 수 없습니다.'}
        
        comments = self.comment_repo.get_by_post(post_id)
        
        return {
            'success': True,
            'post': {
                'id': post.id,
                'title': post.title,
                'content': post.content,
                'author': {
                    'id': post.author.id,
                    'username': post.author.username
                },
                'created_at': post.created_at,
                'updated_at': post.updated_at,
                'view_count': post.view_count,
                'tags': [tag.name for tag in post.tags],
                'comments': [
                    {
                        'id': comment.id,
                        'content': comment.content,
                        'author': {
                            'id': comment.author.id,
                            'username': comment.author.username
                        },
                        'created_at': comment.created_at
                    }
                    for comment in comments
                ]
            }
        }
    
    def get_posts(self, page: int = 1, per_page: int = 10) -> Dict[str, Any]:
        result = self.post_repo.get_all(page, per_page, with_author=True, with_tags=True)
        
        posts = [
            {
                'id': post.id,
                'title': post.title,
                'author': post.author.username,
                'created_at': post.created_at,
                'view_count': post.view_count,
                'tags': [tag.name for tag in post.tags]
            }
            for post in result['posts']
        ]
        
        return {
            'success': True,
            'posts': posts,
            'pagination': {
                'page': result['page'],
                'per_page': result['per_page'],
                'total': result['total'],
                'total_pages': result['total_pages'],
                'has_next': result['has_next'],
                'has_prev': result['has_prev']
            }
        }
    
    def get_posts_by_tag(self, tag_name: str, page: int = 1, per_page: int = 10) -> Dict[str, Any]:
        result = self.post_repo.get_by_tag(tag_name, page, per_page)
        
        posts = [
            {
                'id': post.id,
                'title': post.title,
                'author': post.author.username if post.author else 'Unknown',
                'created_at': post.created_at,
                'view_count': post.view_count
            }
            for post in result['posts']
        ]
        
        return {
            'success': True,
            'tag_name': tag_name,
            'posts': posts,
            'pagination': {
                'page': result['page'],
                'per_page': result['per_page'],
                'total': result['total'],
                'total_pages': result['total_pages'],
                'has_next': result['has_next'],
                'has_prev': result['has_prev']
            }
        }
    
    def update_post(self, post_id: int, user_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
        # 게시글 존재 확인
        post = self.post_repo.get_by_id(post_id)
        if not post:
            return {'success': False, 'message': '게시글을 찾을 수 없습니다.'}
        
        # 권한 확인
        if post.user_id != user_id:
            return {'success': False, 'message': '게시글을 수정할 권한이 없습니다.'}
        
        # 게시글 업데이트
        updated_post = self.post_repo.update(post_id, data)
        
        return {
            'success': True,
            'post': {
                'id': updated_post.id,
                'title': updated_post.title,
                'updated_at': updated_post.updated_at
            }
        }
    
    def delete_post(self, post_id: int, user_id: int) -> Dict[str, Any]:
        # 게시글 존재 확인
        post = self.post_repo.get_by_id(post_id)
        if not post:
            return {'success': False, 'message': '게시글을 찾을 수 없습니다.'}
        
        # 권한 확인
        if post.user_id != user_id:
            return {'success': False, 'message': '게시글을 삭제할 권한이 없습니다.'}
        
        # 게시글 삭제
        success = self.post_repo.delete(post_id)
        
        return {
            'success': success,
            'message': '게시글이 삭제되었습니다.' if success else '게시글 삭제에 실패했습니다.'
        }
    
    def add_comment(self, post_id: int, user_id: int, content: str) -> Dict[str, Any]:
        # 게시글 존재 확인
        post = self.post_repo.get_by_id(post_id)
        if not post:
            return {'success': False, 'message': '게시글을 찾을 수 없습니다.'}
        
        # 댓글 추가
        comment = self.comment_repo.create(content, user_id, post_id)
        
        return {
            'success': True,
            'comment': {
                'id': comment.id,
                'content': comment.content,
                'created_at': comment.created_at
            }
        }
    
    def delete_comment(self, comment_id: int, user_id: int) -> Dict[str, Any]:
        # 댓글 존재 확인 및 권한 확인은 실제 구현에서 추가
        
        # 댓글 삭제
        success = self.comment_repo.delete(comment_id)
        
        return {
            'success': success,
            'message': '댓글이 삭제되었습니다.' if success else '댓글 삭제에 실패했습니다.'
        }
    
    def get_popular_tags(self, limit: int = 10) -> Dict[str, Any]:
        tags = self.tag_repo.get_popular_tags(limit)
        
        return {
            'success': True,
            'tags': tags
        }

from database import engine, get_db_session
from models import Base
from services import UserService, BlogService

def init_db():
    """데이터베이스 초기화"""
    Base.metadata.create_all(bind=engine)
    print("데이터베이스가 초기화되었습니다.")

def demo():
    """간단한 데모 실행"""
    session = get_db_session()
    
    # 서비스 인스턴스 생성
    user_service = UserService(session)
    blog_service = BlogService(session)
    
    # 사용자 등록
    result = user_service.register("admin", "admin@example.com", "password123")
    if result['success']:
        user_id = result['user']['id']
        print(f"사용자가 등록되었습니다: {result['user']['username']}")
    else:
        print(f"사용자 등록 실패: {result['message']}")
        return
    
    # 게시글 작성
    post_result = blog_service.create_post(
        title="SQLAlchemy 시작하기",
        content="SQLAlchemy는 파이썬에서 가장 인기 있는 ORM 라이브러리입니다...",
        user_id=user_id,
        tags=["Python", "SQLAlchemy", "Database"]
    )
    
    if post_result['success']:
        post_id = post_result['post']['id']
        print(f"게시글이 작성되었습니다: {post_result['post']['title']}")
        
        # 댓글 작성
        comment_result = blog_service.add_comment(
            post_id=post_id,
            user_id=user_id,
            content="좋은 게시글입니다!"
        )
        
        if comment_result['success']:
            print("댓글이 작성되었습니다.")
        
        # 게시글 조회
        post_detail = blog_service.get_post(post_id)
        if post_detail['success']:
            post = post_detail['post']
            print(f"\n게시글 상세:")
            print(f"제목: {post['title']}")
            print(f"작성자: {post['author']['username']}")
            print(f"태그: {', '.join(post['tags'])}")
            print(f"조회수: {post['view_count']}")
            print(f"댓글 수: {len(post['comments'])}")
    
    # 인기 태그 조회
    tags_result = blog_service.get_popular_tags()
    if tags_result['success']:
        print("\n인기 태그:")
        for tag in tags_result['tags']:
            print(f"- {tag['name']} ({tag['post_count']}개 게시글)")

if __name__ == "__main__":
    # 데이터베이스 초기화
    init_db()
    
    # 데모 실행
    demo()

# SQLAlchemy 1.x 스타일
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# SQLAlchemy 2.x 스타일
class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100))
    # Optional 필드
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    
    # 일대다 관계
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = 'posts'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    
    # 다대일 관계
    author: Mapped["User"] = relationship("User", back_populates="posts")

# SQLAlchemy 1.x 스타일
users = session.query(User).filter(User.name == 'Alice').all()

# SQLAlchemy 2.x 스타일
stmt = select(User).where(User.name == 'Alice')
result = session.execute(stmt)
users = result.scalars().all()

from sqlalchemy import select

# 단일 엔티티 조회
stmt = select(User).where(User.id == 1)
result = session.execute(stmt)
user = result.scalar_one()  # 정확히 하나의 결과 반환 (없거나 여러 개면 예외)

# 여러 컬럼 조회
stmt = select(User.id, User.name).where(User.id < 10)
result = session.execute(stmt)
for user_id, user_name in result:
    print(f"ID: {user_id}, Name: {user_name}")

# 집계 함수 사용
from sqlalchemy import func
stmt = select(func.count()).select_from(User)
result = session.execute(stmt)
user_count = result.scalar_one()

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select

# 비동기 엔진 생성
async_engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/dbname',
    echo=True
)

# 비동기 세션 사용
async with AsyncSession(async_engine) as session:
    # 비동기 쿼리 실행
    stmt = select(User).where(User.name == 'Alice')
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()
    
    if user:
        # 데이터 수정
        user.email = 'new_email@example.com'
        
        # 변경사항 커밋
        await session.commit()

from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import ForeignKey
from typing import List, Optional, Set

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    
    # 컬렉션 관계 (List, Set 등 지정 가능)
    posts: Mapped[List["Post"]] = relationship(back_populates="author")
    # 고유 컬렉션 (중복 없음)
    badges: Mapped[Set["Badge"]] = relationship(back_populates="users")
    
    # 단일 관계 (일대일)
    profile: Mapped["Profile"] = relationship(back_populates="user")
    # 선택적 관계 (nullable)
    current_post: Mapped[Optional["Post"]] = relationship()

class Post(Base):
    __tablename__ = 'posts'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    
    author: Mapped["User"] = relationship(back_populates="posts")

from sqlalchemy import select, union, union_all, except_, intersect, CTE

# UNION 쿼리
stmt1 = select(User.name).where(User.id < 10)
stmt2 = select(User.name).where(User.name.like('A%'))
union_stmt = union(stmt1, stmt2)

result = session.execute(union_stmt)
names = result.scalars().all()

# WITH 절 (CTE)
active_users = select(User.id, User.name) \
    .where(User.is_active == True) \
    .cte('active_users')

recent_posts = select(Post.id, Post.title, Post.user_id) \
    .where(Post.created_at > datetime(2025, 1, 1)) \
    .cte('recent_posts')

# CTE 조합
stmt = select(
        active_users.c.name,
        func.count(recent_posts.c.id).label('post_count')
    ) \
    .join(recent_posts, active_users.c.id == recent_posts.c.user_id) \
    .group_by(active_users.c.name) \
    .order_by(func.count(recent_posts.c.id).desc())

result = session.execute(stmt)
for name, post_count in result:
    print(f"{name}: {post_count}개 게시글")

# Alembic 설치
pip install alembic==1.12.0

# 초기화
alembic init migrations

# 마이그레이션 생성
alembic revision --autogenerate -m "Create users and posts tables"

# 마이그레이션 적용
alembic upgrade head

# 특정 버전으로 롤백
alembic downgrade -1

from sqlalchemy import JSON, UUID, ARRAY, Enum
import uuid
import enum

class UserRole(enum.Enum):
    ADMIN = "admin"
    EDITOR = "editor"
    USER = "user"

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(50))
    role: Mapped[UserRole] = mapped_column(Enum(UserRole))
    settings: Mapped[dict] = mapped_column(JSON)
    tags: Mapped[list] = mapped_column(ARRAY(String))  # PostgreSQL 전용
    
    def __init__(self, name, role, settings=None, tags=None):
        self.name = name
        self.role = role
        self.settings = settings or {}
        self.tags = tags or []

pip install alembic

alembic init migrations

# alembic.ini
sqlalchemy.url = sqlite:///blog.db

# migrations/env.py

# 기존 코드...

# 모델 가져오기
from models import Base
target_metadata = Base.metadata

# 기존 코드...

alembic revision --autogenerate -m "Create initial tables"

"""Create initial tables

Revision ID: a1b2c3d4e5f6
Revises: 
Create Date: 2025-03-10 12:34:56.789012

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = 'a1b2c3d4e5f6'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('tags',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(length=50), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('name')
    )
    op.create_index(op.f('ix_tags_name'), 'tags', ['name'], unique=True)
    
    op.create_table('users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('email', sa.String(length=100), nullable=False),
        sa.Column('password_hash', sa.String(length=100), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
        sa.UniqueConstraint('username')
    )
    op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
    
    # 나머지 테이블 생성 코드...
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    # 테이블 삭제 코드...
    op.drop_index(op.f('ix_users_username'), table_name='users')
    op.drop_table('users')
    op.drop_index(op.f('ix_tags_name'), table_name='tags')
    op.drop_table('tags')
    # ### end Alembic commands ###

alembic upgrade head

alembic upgrade a1b2c3d4e5f6

# 한 단계 앞으로
alembic upgrade +1

# 두 단계 뒤로
alembic downgrade -2

# models.py 수정
class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    password_hash: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    is_active: Mapped[bool] = mapped_column(default=True)
    # 새 필드 추가
    last_login: Mapped[Optional[datetime]] = mapped_column(nullable=True)
    
    # 관계 설정 코드...

alembic revision --autogenerate -m "Add last_login field to User"

"""Add last_login field to User

Revision ID: b2c3d4e5f6g7
Revises: a1b2c3d4e5f6
Create Date: 2025-03-10 13:45:56.789012

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = 'b2c3d4e5f6g7'
down_revision = 'a1b2c3d4e5f6'
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('users', sa.Column('last_login', sa.DateTime(), nullable=True))
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('users', 'last_login')
    # ### end Alembic commands ###

alembic upgrade head

def upgrade():
    # 스키마 변경
    op.add_column('users', sa.Column('full_name', sa.String(length=100), nullable=True))
    
    # 데이터 마이그레이션
    # 바인드를 통해 SQLAlchemy 엔진에 접근
    conn = op.get_bind()
    
    # 기존 데이터 조회
    result = conn.execute(sa.text("SELECT id, username FROM users"))
    
    # 각 사용자의 full_name 업데이트
    for user_id, username in result:
        conn.execute(
            sa.text("UPDATE users SET full_name = :name WHERE id = :id"),
            {"name": f"{username}'s full name", "id": user_id}
        )

# N+1 쿼리 문제 예시
with Session(engine) as session:
    # 사용자 10명 조회 (1번 쿼리)
    stmt = select(User).limit(10)
    result = session.execute(stmt)
    users = result.scalars().all()
    
    # 각 사용자의 게시글 접근 (사용자당 1번 쿼리, 총 10번의 추가 쿼리)
    for user in users:
        print(f"{user.username}의 게시글: {len(user.posts)}개")

# 즉시 로딩으로 해결
from sqlalchemy.orm import selectinload

with Session(engine) as session:
    # 사용자와 게시글을 한 번에 조회 (1번 쿼리 또는 2번 쿼리)
    stmt = select(User).options(selectinload(User.posts)).limit(10)
    result = session.execute(stmt)
    users = result.scalars().all()
    
    # 추가 쿼리 없이 게시글 접근
    for user in users:
        print(f"{user.username}의 게시글: {len(user.posts)}개")

from sqlalchemy.orm import selectinload, joinedload, subqueryload, raiseload

with Session(engine) as session:
    # 다대일 관계에는 joinedload가 효율적
    stmt = select(Post).options(joinedload(Post.author)).limit(20)
    
    # 일대다 관계에는 selectinload가 효율적
    stmt = select(User).options(selectinload(User.posts)).limit(10)
    
    # 대량의 데이터를 로드할 때는 subqueryload 고려
    stmt = select(Tag).options(subqueryload(Tag.posts)).all()
    
    # 특정 관계는 로드하지 않도록 설정
    stmt = select(User).options(raiseload(User.posts))

from sqlalchemy import update, delete

with Session(engine) as session:
    # 개별 객체 업데이트 (비효율적)
    users = session.execute(select(User).where(User.is_active == False)).scalars().all()
    for user in users:
        user.is_active = True
    session.commit()
    
    # 벌크 업데이트 (효율적)
    stmt = update(User).where(User.is_active == False).values(is_active=True)
    session.execute(stmt)
    session.commit()
    
    # 벌크 삭제
    stmt = delete(Post).where(Post.created_at < datetime(2023, 1, 1))
    result = session.execute(stmt)
    session.commit()
    print(f"{result.rowcount}개의 게시글이 삭제되었습니다.")

from sqlalchemy import Index

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    
    # 테이블 정의 후 인덱스 추가
    __table_args__ = (
        # 복합 인덱스
        Index('idx_user_created_username', created_at, username),
        # 부분 인덱스 (PostgreSQL)
        Index('idx_user_active', 'is_active', postgresql_where=is_active == True),
    )

# 웹 애플리케이션에서의 세션 관리 예시
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool
from threading import local

# 엔진 생성 시 풀링 설정
engine = create_engine(
    'postgresql://user:pass@localhost/dbname',
    poolclass=QueuePool,
    pool_size=10,  # 기본 연결 수
    max_overflow=20,  # 추가로 생성 가능한 최대 연결 수
    pool_timeout=30,  # 연결 대기 시간 (초)
    pool_recycle=3600  # 연결 재활용 시간 (초)
)

# 스레드 로컬 세션 팩토리
Session = scoped_session(
    sessionmaker(bind=engine, expire_on_commit=False)
)

# 요청마다 세션 생성 및 정리
def handle_request():
    try:
        # 세션 얻기
        session = Session()
        
        # 작업 수행
        # ...
        
        # 변경사항 커밋
        session.commit()
    except:
        # 오류 발생 시 롤백
        session.rollback()
        raise
    finally:
        # 세션 정리
        session.close()
        # 스레드 로컬 세션 제거
        Session.remove()

from sqlalchemy import select, func

with Session(engine) as session:
    # 필요한 컬럼만 선택 (전체 객체를 로드하지 않음)
    stmt = select(User.id, User.username).where(User.is_active == True)
    result = session.execute(stmt)
    
    # LIMIT 사용
    stmt = select(Post).order_by(Post.created_at.desc()).limit(10)
    
    # 집계 함수를 데이터베이스에서 처리
    stmt = select(
            func.date_trunc('month', Post.created_at).label('month'),
            func.count().label('post_count')
        ) \
        .group_by('month') \
        .order_by('month')
    
    # 복잡한 조건은 데이터베이스에서 처리
    stmt = select(User) \
        .where(
            (User.created_at > datetime(2024, 1, 1)) &
            (User.email.like('%@example.com')) &
            (User.is_active == True)
        )

import redis
import json
import pickle
from functools import wraps

# Redis 연결
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 캐싱 데코레이터
def cache_query(expire_seconds=3600):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 캐시 키 생성
            key = f"cache:{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # 캐시에서 결과 조회
            cached_result = redis_client.get(key)
            if cached_result:
                return pickle.loads(cached_result)
            
            # 캐시에 없으면 함수 실행
            result = func(*args, **kwargs)
            
            # 결과 캐싱
            redis_client.setex(key, expire_seconds, pickle.dumps(result))
            
            return result
        return wrapper
    return decorator

# 사용 예시
@cache_query(expire_seconds=300)  # 5분 캐싱
def get_popular_posts(session, limit=10):
    stmt = select(Post) \
        .order_by(Post.view_count.desc()) \
        .limit(limit)
    
    result = session.execute(stmt)
    return result.scalars().all()

# 함수 호출
with Session(engine) as session:
    popular_posts = get_popular_posts(session)  # 첫 호출: DB 쿼리 실행
    popular_posts = get_popular_posts(session)  # 두 번째 호출: 캐시에서 가져옴

from sqlalchemy import select

with Session(engine) as session:
    # 모든 데이터를 한 번에 로드 (메모리 문제 가능성)
    all_users = session.execute(select(User)).scalars().all()
    
    # 대신 청크(chunk) 단위로 처리
    stmt = select(User).order_by(User.id)
    
    # 스트리밍 방식으로 처리
    result = session.execute(stmt)
    chunk_size = 1000
    users_chunk = []
    
    for i, user in enumerate(result.scalars()):
        users_chunk.append(user)
        
        # 청크 크기에 도달하면 처리
        if len(users_chunk) >= chunk_size:
            process_users_chunk(users_chunk)
            users_chunk = []
    
    # 남은 데이터 처리
    if users_chunk:
        process_users_chunk(users_chunk)
    
    # 또는 yield_per 사용 (일부 데이터베이스에서만 지원)
    result = session.execute(stmt.execution_options(yield_per=1000))
    for user in result.scalars():
        process_user(user)

# SQLAlchemy 로깅 활성화
import logging

logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# 또는 특정 엔진에만 로깅 설정
engine = create_engine('sqlite:///example.db', echo=True)

# 쿼리 실행 시간 측정
import time

def measure_query_time(session, query_func):
    start_time = time.time()
    result = query_func(session)
    end_time = time.time()
    
    print(f"쿼리 실행 시간: {end_time - start_time:.4f}초")
    return result

# 사용 예시
with Session(engine) as session:
    result = measure_query_time(session, lambda s: s.execute(select(User)).scalars().all())