# 전통적인 방식 - SQL 직접 작성
cursor.execute("SELECT * FROM users WHERE age > 20")
results = cursor.fetchall()

# 데이터 삽입도 복잡해
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)", 
               (name, email, age))

# ORM 방식 - Python 객체로 작업
users = session.query(User).filter(User.age > 20).all()

# 데이터 삽입도 간단해
new_user = User(name=name, email=email, age=age)
session.add(new_user)
session.commit()

# 기본 SQLAlchemy 설치
pip install sqlalchemy

# 특정 데이터베이스 드라이버도 함께 설치
pip install sqlalchemy pymysql  # MySQL용
pip install sqlalchemy psycopg2-binary  # PostgreSQL용
# SQLite는 Python 기본 내장이라 별도 설치 불필요!

import sqlalchemy
print(sqlalchemy.__version__)  # 2.0 이상이면 최신 버전!

# 기본 형식
dialect+driver://username:password@host:port/database

# 실제 예시들
# SQLite (파일 기반)
sqlite:///mydb.db

# MySQL
mysql+pymysql://user:password@localhost:3306/mydb

# PostgreSQL
postgresql+psycopg2://user:password@localhost:5432/mydb

# 메모리 내 SQLite (테스트용으로 좋아!)
sqlite:///:memory:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 1. 엔진 생성 (데이터베이스 연결)
engine = create_engine('sqlite:///example.db', echo=True)

# echo=True는 실행되는 SQL을 콘솔에 출력해줘 (디버깅용)
# 실제 운영에서는 echo=False로 설정하는 게 좋아

# 2. 세션 팩토리 생성
SessionLocal = sessionmaker(bind=engine)

# 3. 세션 인스턴스 생성
session = SessionLocal()

print("데이터베이스 연결 성공! 🎉")

# MySQL 연결
engine = create_engine(
    'mysql+pymysql://myuser:mypassword@localhost:3306/mydb',
    pool_size=10,  # 연결 풀 크기
    max_overflow=20,  # 최대 추가 연결 수
    pool_pre_ping=True  # 연결 유효성 자동 체크
)

# PostgreSQL 연결
engine = create_engine(
    'postgresql+psycopg2://myuser:mypassword@localhost:5432/mydb',
    pool_size=10,
    echo=False
)

# 연결 풀 세밀하게 설정하기
from sqlalchemy.pool import QueuePool

engine = create_engine(
    'mysql+pymysql://user:pass@localhost/mydb',
    poolclass=QueuePool,
    pool_size=20,  # 기본 20개 연결 유지
    max_overflow=30,  # 최대 50개까지 가능
    pool_timeout=30,  # 30초 대기
    pool_recycle=3600,  # 1시간마다 연결 재생성
    pool_pre_ping=True  # 사용 전 연결 체크
)

from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

# Base 클래스 생성
Base = declarative_base()

# User 모델 정의
class User(Base):
    __tablename__ = 'users'  # 실제 테이블 이름
    
    # 컬럼 정의
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 객체를 출력할 때 보기 좋게 (선택사항)
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"

# 모든 테이블 생성
Base.metadata.create_all(engine)

# 특정 테이블만 생성
Base.metadata.tables['users'].create(engine)

# 테이블 삭제 (조심해서 사용!)
Base.metadata.drop_all(engine)

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Table
from sqlalchemy.orm import relationship
from datetime import datetime

Base = declarative_base()

# 다대다 관계를 위한 연결 테이블
post_tags = Table('post_tags', Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id')),
    Column('tag_id', Integer, ForeignKey('tags.id'))
)

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    bio = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 관계 설정: 한 사용자는 여러 포스트를 가질 수 있음
    posts = relationship('Post', back_populates='author')
    comments = relationship('Comment', back_populates='user')

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, onupdate=datetime.utcnow)
    view_count = Column(Integer, default=0)
    
    # 관계 설정
    author = relationship('User', back_populates='posts')
    comments = relationship('Comment', back_populates='post')
    tags = relationship('Tag', secondary=post_tags, back_populates='posts')

class Comment(Base):
    __tablename__ = 'comments'
    
    id = Column(Integer, primary_key=True)
    content = Column(Text, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    post_id = Column(Integer, ForeignKey('posts.id'))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    user = relationship('User', back_populates='comments')
    post = relationship('Post', back_populates='comments')

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    
    posts = relationship('Post', secondary=post_tags, back_populates='tags')

from sqlalchemy.orm import Session

# 세션 생성
session = Session(engine)

# 1. 단일 사용자 추가
new_user = User(
    username='john_doe',
    email='john@example.com',
    password='hashed_password_here',
    bio='Python 개발자입니다!'
)

session.add(new_user)
session.commit()  # 실제로 DB에 저장

print(f"새 사용자 생성! ID: {new_user.id}")  # ID가 자동으로 할당됨

# 2. 여러 개 한 번에 추가
users = [
    User(username='alice', email='alice@example.com', password='pass1'),
    User(username='bob', email='bob@example.com', password='pass2'),
    User(username='charlie', email='charlie@example.com', password='pass3')
]

session.add_all(users)
session.commit()

print(f"{len(users)}명의 사용자 추가 완료!")

# 1. 모든 사용자 가져오기
all_users = session.query(User).all()
print(f"총 {len(all_users)}명의 사용자")

# 2. 첫 번째 사용자만 가져오기
first_user = session.query(User).first()
print(f"첫 번째 사용자: {first_user.username}")

# 3. ID로 특정 사용자 찾기
user = session.query(User).get(1)  # ID가 1인 사용자
# 또는
user = session.query(User).filter(User.id == 1).first()

# 4. 조건으로 필터링
active_users = session.query(User).filter(User.is_active == True).all()

# 5. 여러 조건 (AND)
user = session.query(User).filter(
    User.username == 'john_doe',
    User.is_active == True
).first()

# 6. OR 조건
from sqlalchemy import or_
users = session.query(User).filter(
    or_(User.username == 'alice', User.username == 'bob')
).all()

# 7. LIKE 검색 (부분 일치)
users = session.query(User).filter(User.email.like('%@example.com')).all()

# 8. IN 조건
usernames = ['alice', 'bob', 'charlie']
users = session.query(User).filter(User.username.in_(usernames)).all()

# 9. 정렬
users = session.query(User).order_by(User.created_at.desc()).all()  # 최신순
users = session.query(User).order_by(User.username.asc()).all()  # 이름순

# 10. 개수 제한
recent_users = session.query(User).order_by(User.created_at.desc()).limit(10).all()

# 11. 개수 세기
user_count = session.query(User).count()
print(f"총 사용자 수: {user_count}")

# 12. 페이지네이션
page = 1
per_page = 20
users = session.query(User)\
    .order_by(User.created_at.desc())\
    .offset((page - 1) * per_page)\
    .limit(per_page)\
    .all()

# 13. 특정 컬럼만 선택
usernames = session.query(User.username, User.email).all()
for username, email in usernames:
    print(f"{username}: {email}")

# 14. 집계 함수
from sqlalchemy import func
avg_view_count = session.query(func.avg(Post.view_count)).scalar()
max_views = session.query(func.max(Post.view_count)).scalar()

# 15. 그룹화
from sqlalchemy import func
post_counts = session.query(
    User.username,
    func.count(Post.id).label('post_count')
).join(Post).group_by(User.username).all()

for username, count in post_counts:
    print(f"{username}: {count}개의 포스트")

# 1. 단일 객체 수정
user = session.query(User).filter(User.username == 'john_doe').first()
if user:
    user.email = 'newemail@example.com'
    user.bio = '업데이트된 소개입니다!'
    session.commit()
    print("사용자 정보 업데이트 완료!")

# 2. 여러 객체 한 번에 수정
session.query(User).filter(User.is_active == False).update({
    'is_active': True
})
session.commit()
print("비활성 사용자들을 모두 활성화했습니다!")

# 3. 조건부 업데이트
session.query(Post).filter(Post.view_count < 10).update({
    'view_count': Post.view_count + 1
})
session.commit()

# 4. 관계된 객체 수정
post = session.query(Post).first()
post.title = "새로운 제목"
post.content = "새로운 내용"
# 태그 추가
new_tag = Tag(name='Python')
post.tags.append(new_tag)
session.commit()

# 1. 단일 객체 삭제
user = session.query(User).filter(User.username == 'test_user').first()
if user:
    session.delete(user)
    session.commit()
    print("사용자 삭제 완료!")

# 2. 조건에 맞는 여러 객체 삭제
deleted_count = session.query(Post).filter(Post.view_count == 0).delete()
session.commit()
print(f"{deleted_count}개의 포스트 삭제!")

# 3. 관계된 객체 처리
# CASCADE 설정이 되어 있으면 자동으로 관련 데이터도 삭제됨
user = session.query(User).get(1)
session.delete(user)  # 이 사용자의 포스트와 댓글도 함께 삭제될 수 있음
session.commit()

try:
    # 여러 작업을 하나의 트랜잭션으로
    new_user = User(username='newuser', email='new@example.com')
    session.add(new_user)
    
    new_post = Post(title='첫 포스트', content='내용', author=new_user)
    session.add(new_post)
    
    # 모든 작업이 성공하면 커밋
    session.commit()
    print("모든 작업 성공!")
    
except Exception as e:
    # 에러 발생 시 롤백 (모든 변경사항 취소)
    session.rollback()
    print(f"에러 발생! 롤백합니다: {e}")
    
finally:
    # 세션 정리
    session.close()

from contextlib import contextmanager

@contextmanager
def get_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# 사용
with get_session() as session:
    user = User(username='contextuser', email='context@example.com')
    session.add(user)
    # commit과 close가 자동으로 처리됨!

# 사용자의 모든 포스트 가져오기
user = session.query(User).filter(User.username == 'john_doe').first()
user_posts = user.posts  # relationship 덕분에 간단!

print(f"{user.username}의 포스트:")
for post in user_posts:
    print(f"- {post.title}")

# 반대 방향: 포스트의 작성자
post = session.query(Post).first()
author = post.author
print(f"작성자: {author.username}")

# 새 포스트 추가
new_post = Post(
    title="SQLAlchemy 튜토리얼",
    content="ORM은 정말 편리해요!",
    author=user  # 관계 설정
)
session.add(new_post)
session.commit()

# 또는 이렇게도 가능
user.posts.append(Post(title="또 다른 포스트", content="내용"))
session.commit()

# 태그 생성
python_tag = Tag(name='Python')
orm_tag = Tag(name='ORM')
db_tag = Tag(name='Database')

session.add_all([python_tag, orm_tag, db_tag])

# 포스트에 태그 추가
post = session.query(Post).first()
post.tags.extend([python_tag, orm_tag, db_tag])
session.commit()

# 포스트의 모든 태그 조회
print(f"{post.title}의 태그:")
for tag in post.tags:
    print(f"- {tag.name}")

# 특정 태그를 가진 모든 포스트 조회
python_posts = session.query(Post).join(Post.tags).filter(Tag.name == 'Python').all()
print(f"Python 태그가 있는 포스트: {len(python_posts)}개")

# 태그 제거
post.tags.remove(db_tag)
session.commit()

from sqlalchemy.orm import joinedload, selectinload, subqueryload

# 1. Lazy Loading (기본) - N+1 문제 발생!
users = session.query(User).all()
for user in users:  # 각 사용자마다 추가 쿼리 발생
    print(f"{user.username}: {len(user.posts)}개 포스트")

# 2. Joined Load - JOIN으로 한 번에 가져오기
users = session.query(User).options(joinedload(User.posts)).all()
for user in users:  # 추가 쿼리 없음!
    print(f"{user.username}: {len(user.posts)}개 포스트")

# 3. Select IN Load - 별도 쿼리지만 효율적
users = session.query(User).options(selectinload(User.posts)).all()

# 4. 중첩된 관계도 가능
posts = session.query(Post).options(
    joinedload(Post.author),
    selectinload(Post.comments).joinedload(Comment.user),
    selectinload(Post.tags)
).all()

# 각 포스트의 작성자, 댓글(+댓글 작성자), 태그를 모두 미리 로드!

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    
    # CASCADE 옵션 설정
    posts = relationship(
        'Post',
        back_populates='author',
        cascade='all, delete-orphan'  # 사용자 삭제 시 포스트도 삭제
    )

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    author_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'))
    
    author = relationship('User', back_populates='posts')

from sqlalchemy import func

# 평균 이상의 조회수를 가진 포스트 찾기
avg_views = session.query(func.avg(Post.view_count)).scalar_subquery()
popular_posts = session.query(Post).filter(Post.view_count > avg_views).all()

# 서브쿼리를 변수로 만들기
subq = session.query(
    Post.author_id,
    func.count(Post.id).label('post_count')
).group_by(Post.author_id).subquery()

# 포스트를 3개 이상 쓴 사용자 찾기
active_authors = session.query(User).join(
    subq, User.id == subq.c.author_id
).filter(subq.c.post_count >= 3).all()

print(f"활발한 작성자: {len(active_authors)}명")

# INNER JOIN
results = session.query(User, Post).join(Post).all()
for user, post in results:
    print(f"{user.username} - {post.title}")

# LEFT OUTER JOIN (포스트가 없는 사용자도 포함)
results = session.query(User).outerjoin(Post).all()

# 여러 테이블 JOIN
results = session.query(Post, User, Comment).join(
    User, Post.author_id == User.id
).join(
    Comment, Post.id == Comment.post_id
).all()

# 조건부 JOIN
from sqlalchemy import and_
results = session.query(User).join(
    Post,
    and_(
        Post.author_id == User.id,
        Post.view_count > 100
    )
).all()

from sqlalchemy import func, desc

# 사용자별 포스트 개수
post_counts = session.query(
    User.username,
    func.count(Post.id).label('post_count')
).join(Post).group_by(User.username).all()

for username, count in post_counts:
    print(f"{username}: {count}개")

# HAVING 절 사용 (그룹화 후 필터링)
active_users = session.query(
    User.username,
    func.count(Post.id).label('post_count')
).join(Post).group_by(User.username).having(
    func.count(Post.id) > 5
).all()

# 여러 집계 함수 동시 사용
stats = session.query(
    func.count(Post.id).label('total_posts'),
    func.avg(Post.view_count).label('avg_views'),
    func.max(Post.view_count).label('max_views'),
    func.min(Post.view_count).label('min_views')
).first()

print(f"총 포스트: {stats.total_posts}")
print(f"평균 조회수: {stats.avg_views:.2f}")
print(f"최대 조회수: {stats.max_views}")
print(f"최소 조회수: {stats.min_views}")

from sqlalchemy import func, over

# 각 사용자의 포스트에 순위 매기기
ranked_posts = session.query(
    Post.title,
    Post.view_count,
    User.username,
    func.row_number().over(
        partition_by=Post.author_id,
        order_by=Post.view_count.desc()
    ).label('rank')
).join(User).all()

for post in ranked_posts:
    print(f"{post.username}의 {post.rank}위: {post.title} ({post.view_count} views)")

# 누적 합계
cumulative = session.query(
    Post.created_at,
    Post.view_count,
    func.sum(Post.view_count).over(
        order_by=Post.created_at
    ).label('cumulative_views')
).order_by(Post.created_at).all()

from sqlalchemy import func

# PostgreSQL의 to_tsvector와 to_tsquery 사용
search_query = 'python & sqlalchemy'
results = session.query(Post).filter(
    func.to_tsvector('english', Post.content).match(search_query)
).all()

# 검색 결과 랭킹
ranked_results = session.query(
    Post,
    func.ts_rank(
        func.to_tsvector('english', Post.content),
        func.to_tsquery('english', search_query)
    ).label('rank')
).filter(
    func.to_tsvector('english', Post.content).match(search_query)
).order_by(desc('rank')).all()

# UNION - 두 쿼리 결과 합치기
query1 = session.query(User.username).filter(User.is_active == True)
query2 = session.query(User.username).filter(User.created_at > '2024-01-01')
combined = query1.union(query2).all()

# UNION ALL (중복 포함)
combined_all = query1.union_all(query2).all()

# INTERSECT - 교집합
common = query1.intersect(query2).all()

# EXCEPT - 차집합
difference = query1.except_(query2).all()

from sqlalchemy import Index

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))
    created_at = Column(DateTime)
    
    # 단일 컬럼 인덱스
    __table_args__ = (
        Index('idx_username', 'username'),
        Index('idx_email', 'email'),
        Index('idx_created_at', 'created_at'),
    )

# 복합 인덱스
class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    author_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime)
    is_published = Column(Boolean)
    
    # 여러 컬럼을 함께 인덱싱
    __table_args__ = (
        Index('idx_author_created', 'author_id', 'created_at'),
        Index('idx_published_created', 'is_published', 'created_at'),
    )

# 1. Bulk Insert - 빠른 삽입
users_data = [
    {'username': f'user{i}', 'email': f'user{i}@example.com'}
    for i in range(10000)
]

session.bulk_insert_mappings(User, users_data)
session.commit()
print("10,000명 사용자 추가 완료!")

# 2. Bulk Update - 빠른 업데이트
updates = [
    {'id': 1, 'view_count': 100},
    {'id': 2, 'view_count': 200},
    {'id': 3, 'view_count': 300},
]

session.bulk_update_mappings(Post, updates)
session.commit()

# 3. 청크 단위로 처리 (메모리 효율적)
def process_in_chunks(query, chunk_size=1000):
    offset = 0
    while True:
        chunk = query.limit(chunk_size).offset(offset).all()
        if not chunk:
            break
        
        for item in chunk:
            # 처리 로직
            pass
        
        offset += chunk_size
        session.commit()  # 청크마다 커밋

# 사용
query = session.query(Post)
process_in_chunks(query)

# 4. 스트리밍 방식 (yield_per)
for user in session.query(User).yield_per(100):
    # 100개씩 가져와서 처리
    process_user(user)

import time
from sqlalchemy import event
from sqlalchemy.engine import Engine

# 쿼리 실행 시간 측정
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop(-1)
    if total > 0.1:  # 0.1초 이상 걸린 쿼리만 로깅
        print(f"느린 쿼리 발견! {total:.3f}초")
        print(f"SQL: {statement[:100]}...")

# EXPLAIN으로 쿼리 실행 계획 보기
from sqlalchemy import text

query = session.query(User).filter(User.username == 'john')
explain = session.execute(
    text(f"EXPLAIN {query.statement.compile(compile_kwargs={'literal_binds': True})}")
)
for row in explain:
    print(row)

from functools import lru_cache
import redis

# Python 내장 캐시 (간단한 경우)
@lru_cache(maxsize=128)
def get_user_by_id(user_id):
    return session.query(User).get(user_id)

# Redis 캐시 (분산 환경)
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_user_cached(user_id):
    # 캐시 확인
    cached = redis_client.get(f'user:{user_id}')
    if cached:
        return json.loads(cached)
    
    # DB 조회
    user = session.query(User).get(user_id)
    if user:
        # 캐시 저장 (10분 TTL)
        redis_client.setex(
            f'user:{user_id}',
            600,
            json.dumps({'id': user.id, 'username': user.username})
        )
    return user

# 캐시 무효화
def update_user(user_id, **kwargs):
    user = session.query(User).get(user_id)
    for key, value in kwargs.items():
        setattr(user, key, value)
    session.commit()
    
    # 캐시 삭제
    redis_client.delete(f'user:{user_id}')

# ❌ 나쁜 예 - SQL Injection 취약!
username = request.form['username']
query = f"SELECT * FROM users WHERE username = '{username}'"
# 공격자가 username에 "admin' OR '1'='1" 입력하면 모든 사용자 조회됨!

# ✅ 좋은 예 - SQLAlchemy가 자동으로 이스케이프 처리
username = request.form['username']
user = session.query(User).filter(User.username == username).first()

# ✅ 파라미터 바인딩 사용
from sqlalchemy import text
result = session.execute(
    text("SELECT * FROM users WHERE username = :username"),
    {"username": username}
)

# ❌ 절대 하지 말 것!
session.execute(text(f"SELECT * FROM users WHERE username = '{username}'"))

from werkzeug.security import generate_password_hash, check_password_hash

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    password_hash = Column(String(255))  # 해시된 비밀번호 저장
    
    def set_password(self, password):
        """비밀번호를 해시화해서 저장"""
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        """비밀번호 확인"""
        return check_password_hash(self.password_hash, password)

# 사용
new_user = User(username='john')
new_user.set_password('my_secret_password')  # 해시화되어 저장
session.add(new_user)
session.commit()

# 로그인 시
user = session.query(User).filter(User.username == 'john').first()
if user and user.check_password('my_secret_password'):
    print("로그인 성공!")
else:
    print("비밀번호가 틀렸습니다.")

from enum import Enum

class UserRole(Enum):
    ADMIN = 'admin'
    MODERATOR = 'moderator'
    USER = 'user'

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    role = Column(String(20), default=UserRole.USER.value)
    
    def has_permission(self, required_role):
        """권한 체크"""
        role_hierarchy = {
            UserRole.ADMIN: 3,
            UserRole.MODERATOR: 2,
            UserRole.USER: 1
        }
        user_level = role_hierarchy.get(UserRole(self.role), 0)
        required_level = role_hierarchy.get(required_role, 0)
        return user_level >= required_level

# 사용
def delete_post(user, post):
    if not user.has_permission(UserRole.MODERATOR):
        raise PermissionError("권한이 없습니다!")
    
    session.delete(post)
    session.commit()

class AuditLog(Base):
    __tablename__ = 'audit_logs'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    action = Column(String(50))  # 'create', 'update', 'delete'
    table_name = Column(String(50))
    record_id = Column(Integer)
    old_value = Column(Text)  # JSON 형태로 저장
    new_value = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)
    ip_address = Column(String(45))

def log_action(user_id, action, table_name, record_id, old_value=None, new_value=None):
    """작업 로그 기록"""
    log = AuditLog(
        user_id=user_id,
        action=action,
        table_name=table_name,
        record_id=record_id,
        old_value=json.dumps(old_value) if old_value else None,
        new_value=json.dumps(new_value) if new_value else None
    )
    session.add(log)
    session.commit()

# 사용
user = session.query(User).get(1)
old_email = user.email
user.email = 'newemail@example.com'
session.commit()

log_action(
    user_id=current_user.id,
    action='update',
    table_name='users',
    record_id=user.id,
    old_value={'email': old_email},
    new_value={'email': user.email}
)

from cryptography.fernet import Fernet
import os

# 암호화 키 생성 (환경 변수로 관리)
ENCRYPTION_KEY = os.getenv('ENCRYPTION_KEY', Fernet.generate_key())
cipher_suite = Fernet(ENCRYPTION_KEY)

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    _ssn = Column('ssn', LargeBinary)  # 주민번호 암호화 저장
    
    @property
    def ssn(self):
        """복호화해서 반환"""
        if self._ssn:
            return cipher_suite.decrypt(self._ssn).decode()
        return None
    
    @ssn.setter
    def ssn(self, value):
        """암호화해서 저장"""
        if value:
            self._ssn = cipher_suite.encrypt(value.encode())
        else:
            self._ssn = None

# 사용
user = User(username='john')
user.ssn = '123-45-6789'  # 자동으로 암호화되어 저장
session.add(user)
session.commit()

# 조회
user = session.query(User).first()
print(user.ssn)  # 자동으로 복호화되어 출력

# Alembic 설치
pip install alembic

# 프로젝트 초기화
alembic init alembic

# 디렉토리 구조
# myproject/
#   ├── alembic/
#   │   ├── versions/  # 마이그레이션 파일들
#   │   ├── env.py
#   │   └── script.py.mako
#   ├── alembic.ini
#   └── models.py

# alembic.ini
sqlalchemy.url = sqlite:///./mydb.db
# 또는
sqlalchemy.url = mysql+pymysql://user:pass@localhost/mydb

# env.py에서 모델 import
from myapp.models import Base

target_metadata = Base.metadata

# 자동으로 변경사항 감지해서 마이그레이션 생성
alembic revision --autogenerate -m "Add user table"

# 수동으로 마이그레이션 생성
alembic revision -m "Add custom index"

# 생성된 파일 예시 (versions/xxxx_add_user_table.py)
def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(50), nullable=False),
        sa.Column('email', sa.String(100), nullable=False),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index('idx_username', 'users', ['username'])

def downgrade():
    op.drop_index('idx_username', 'users')
    op.drop_table('users')

# 최신 버전으로 업그레이드
alembic upgrade head

# 특정 버전으로 업그레이드
alembic upgrade ae1027a6acf

# 한 단계 업그레이드
alembic upgrade +1

# 다운그레이드 (롤백)
alembic downgrade -1
alembic downgrade base  # 처음으로

# 현재 버전 확인
alembic current

# 마이그레이션 히스토리 보기
alembic history

# 컬럼 추가
def upgrade():
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))

def downgrade():
    op.drop_column('users', 'phone')

# 컬럼 타입 변경
def upgrade():
    op.alter_column('users', 'bio',
                    existing_type=sa.String(255),
                    type_=sa.Text(),
                    existing_nullable=True)

# 인덱스 추가
def upgrade():
    op.create_index('idx_email', 'users', ['email'], unique=True)

def downgrade():
    op.drop_index('idx_email', 'users')

# 외래 키 추가
def upgrade():
    op.create_foreign_key(
        'fk_posts_author',
        'posts', 'users',
        ['author_id'], ['id'],
        ondelete='CASCADE'
    )

# 데이터 마이그레이션
def upgrade():
    # 기존 데이터 업데이트
    op.execute("UPDATE users SET is_active = true WHERE is_active IS NULL")
    
    # 새 컬럼을 NOT NULL로 변경
    op.alter_column('users', 'is_active', nullable=False)