Day 4 - DynamoDB: 트랜잭션, 조건부 쓰기, TTL의 내부 동작

import boto3
 
dynamodb = boto3.client('dynamodb')
 
# TransactWriteItems: 여러 테이블 원자적 쓰기
response = dynamodb.transact_write_items(
    TransactItems=[
        # 1. 주문 테이블에 새 주문 추가
        {
            'Put': {
                'TableName': 'Orders',
                'Item': {
                    'orderId': {'S': 'O001'},
                    'userId': {'S': 'U001'},
                    'total': {'N': '50000'},
                    'status': {'S': 'PENDING'}
                },
                'ConditionExpression': 'attribute_not_exists(orderId)'
            }
        },
        # 2. 재고 테이블에서 수량 감소 (0 이상인 경우만)
        {
            'Update': {
                'TableName': 'Inventory',
                'Key': {'productId': {'S': 'P001'}},
                'UpdateExpression': 'SET quantity = quantity - :qty',
                'ConditionExpression': 'quantity >= :qty',
                'ExpressionAttributeValues': {':qty': {'N': '1'}}
            }
        },
        # 3. 다른 테이블의 조건만 확인 (쓰기 없음)
        {
            'ConditionCheck': {
                'TableName': 'Users',
                'Key': {'userId': {'S': 'U001'}},
                'ConditionExpression': 'attribute_exists(userId) AND accountStatus = :active',
                'ExpressionAttributeValues': {':active': {'S': 'ACTIVE'}}
            }
        }
    ],
    # 멱등성 키: 같은 토큰으로 재시도 시 중복 실행 방지 (10분 유효)
    ClientRequestToken='unique-tx-id-2026-06-26-001'
)

# 같은 orderId로 두 번 주문이 들어오는 것을 방지
table.put_item(
    Item={'orderId': 'O001', 'total': 50000},
    ConditionExpression='attribute_not_exists(orderId)'
    # orderId 속성이 없을 때만 삽입 (이미 있으면 예외 발생)
)

# 재고가 충분할 때만 차감
table.update_item(
    Key={'productId': 'P001'},
    UpdateExpression='SET quantity = quantity - :qty',
    ConditionExpression='quantity >= :qty',
    ExpressionAttributeValues={':qty': {'N': '1'}}
)

# 버전 번호로 동시 수정 충돌 감지
def update_with_optimistic_lock(table, product_id, new_price, current_version):
    try:
        table.update_item(
            Key={'productId': product_id},
            UpdateExpression='SET price = :p, version = version + :inc',
            ConditionExpression='version = :v',
            ExpressionAttributeValues={
                ':p': new_price,
                ':inc': 1,
                ':v': current_version  # 읽은 버전과 일치해야만 수정
            }
        )
        return True
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            # 다른 프로세스가 먼저 수정함 → 최신 데이터 다시 읽고 재시도
            return False
        raise

# 쿠폰을 가장 먼저 클레임한 사용자만 성공
table.update_item(
    Key={'couponId': 'C001'},
    UpdateExpression='SET claimedBy = :userId, claimedAt = :ts',
    ConditionExpression='attribute_not_exists(claimedBy)',
    ExpressionAttributeValues={
        ':userId': {'S': 'U001'},
        ':ts': {'S': '2026-06-26T10:00:00Z'}
    }
)

# Atomic Counter — 항상 증가 (멱등성 없음)
table.update_item(
    Key={'pageId': 'home'},
    UpdateExpression='ADD viewCount :inc',
    ExpressionAttributeValues={':inc': {'N': '1'}}
)
 
# Conditional Increment — 한도 검사 포함
table.update_item(
    Key={'userId': 'U001'},
    UpdateExpression='SET loginCount = loginCount + :inc',
    ConditionExpression='loginCount < :maxLogins',
    ExpressionAttributeValues={':inc': {'N': '1'}, ':maxLogins': {'N': '5'}}
)

import time
 
# TTL 속성 추가: 24시간 후 만료
expiry_time = int(time.time()) + 86400  # 현재 시간 + 24시간
 
table.put_item(
    Item={
        'sessionId': 'S001',
        'userId': 'U001',
        'data': '...',
        'ttl': expiry_time  # 이 속성명은 테이블 TTL 설정에서 지정
    }
)

# 만료된 항목 필터링이 필요한 경우
import time
 
response = table.query(
    KeyConditionExpression=Key('userId').eq('U001'),
    FilterExpression=Attr('ttl').gt(int(time.time()))
    # TTL이 현재 시간보다 큰 항목만 (유효한 항목)
)

import time
import uuid
 
LOCK_TABLE = 'DistributedLocks'
LOCK_TTL_SECONDS = 30  # 락 최대 보유 시간
 
def acquire_lock(resource_id: str, owner_id: str = None) -> str | None:
    """락 획득. 성공 시 lock_token 반환, 실패 시 None"""
    if owner_id is None:
        owner_id = str(uuid.uuid4())
    
    try:
        dynamodb.put_item(
            TableName=LOCK_TABLE,
            Item={
                'resourceId': {'S': resource_id},
                'ownerId': {'S': owner_id},
                'ttl': {'N': str(int(time.time()) + LOCK_TTL_SECONDS)}
            },
            # 락이 없거나 TTL이 만료된 경우에만 획득
            ConditionExpression='attribute_not_exists(resourceId)'
        )
        return owner_id
    except dynamodb.exceptions.ConditionalCheckFailedException:
        return None  # 다른 프로세스가 락 보유 중
 
def release_lock(resource_id: str, owner_id: str) -> bool:
    """내가 보유한 락만 해제"""
    try:
        dynamodb.delete_item(
            TableName=LOCK_TABLE,
            Key={'resourceId': {'S': resource_id}},
            ConditionExpression='ownerId = :owner',
            ExpressionAttributeValues={':owner': {'S': owner_id}}
        )
        return True
    except dynamodb.exceptions.ConditionalCheckFailedException:
        return False  # 다른 프로세스의 락이거나 이미 만료