Last active
April 21, 2026 03:02
-
-
Save koorukuroo/eb2e4bfc77d2ccdb74dfcd1e51c0c7ec to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import json | |
| import boto3 | |
| import uuid | |
| from decimal import Decimal | |
| # DynamoDB 설정 | |
| dynamodb = boto3.resource('dynamodb') | |
| table = dynamodb.Table('ItemsTable') # 테이블 이름 수정 | |
| # JSON 직렬화용 (Decimal → float) | |
| class DecimalEncoder(json.JSONEncoder): | |
| def default(self, obj): | |
| if isinstance(obj, Decimal): | |
| return float(obj) | |
| return super().default(obj) | |
| # 공통 응답 함수 | |
| def response(status_code, body): | |
| return { | |
| 'statusCode': status_code, | |
| 'headers': {'Content-Type': 'application/json'}, | |
| 'body': json.dumps(body, cls=DecimalEncoder) | |
| } | |
| # 전체 조회 | |
| def get_all_items(): | |
| res = table.scan() | |
| return response(200, res.get('Items', [])) | |
| # 단건 조회 | |
| def get_item(item_id): | |
| res = table.get_item(Key={'id': item_id}) | |
| item = res.get('Item') | |
| if not item: | |
| return response(404, {'error': 'Item not found'}) | |
| return response(200, item) | |
| # 생성 | |
| def create_item(event): | |
| body = json.loads(event.get('body', '{}')) | |
| item = { | |
| 'id': str(uuid.uuid4()), | |
| 'name': body.get('name'), | |
| 'value': body.get('value') | |
| } | |
| table.put_item(Item=item) | |
| return response(201, item) | |
| # 수정 | |
| def update_item(item_id, event): | |
| if not item_id: | |
| return response(400, {'error': 'id 필요'}) | |
| body = json.loads(event.get('body', '{}')) | |
| table.update_item( | |
| Key={'id': item_id}, | |
| UpdateExpression="set #n = :n, #v = :v", | |
| ExpressionAttributeNames={ | |
| '#n': 'name', | |
| '#v': 'value' | |
| }, | |
| ExpressionAttributeValues={ | |
| ':n': body.get('name'), | |
| ':v': body.get('value') | |
| } | |
| ) | |
| return response(200, {'message': 'updated'}) | |
| # 삭제 | |
| def delete_item(item_id): | |
| if not item_id: | |
| return response(400, {'error': 'id 필요'}) | |
| table.delete_item(Key={'id': item_id}) | |
| return response(200, {'message': 'deleted'}) | |
| # 메인 핸들러 | |
| def lambda_handler(event, context): | |
| method = event['httpMethod'] | |
| path_params = event.get('pathParameters') or {} | |
| if method == 'GET' and 'id' not in path_params: | |
| return get_all_items() | |
| elif method == 'GET' and 'id' in path_params: | |
| return get_item(path_params['id']) | |
| elif method == 'POST': | |
| return create_item(event) | |
| elif method == 'PUT': | |
| return update_item(path_params.get('id'), event) | |
| elif method == 'DELETE': | |
| return delete_item(path_params.get('id')) | |
| else: | |
| return response(400, {'error': '지원하지 않는 메서드'}) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment