経緯
最近、AWS Lambdaを使ってデバイス情報のCRUD処理を実装しました。
コードレビューを初めて受けたことで、多くの学びがあったので備忘録としてまとめておきます。
レビュー前のコード🐤
ランタイム:python3.12
import json
import boto3
dynamo = boto3.resource('dynamodb')
table = dynamo.Table('hogehoge')
counter_table = dynamo.Table('deviceidCounter')
kvs = boto3.client('kinesisvideo')
def lambda_handler(event, context):
body = json.loads(event['body'])
device_id = create_device_id()
channel_name = f"webrtc-{device_id}"
kvs.create_signaling_channel(ChannelName=channel_name)
item = {
'deviceId': device_id,
'name': body['name'],
'location': body['location'],
'model': body['model'],
'serialNumber': body['serialNumber']
}
table.put_item(Item=item)
return {
'statusCode': 200,
'body': json.dumps({'message': 'success'})
}
def create_device_id():
res = counter_table.get_item(Key={'counterId': 'counter'})
curr = int(res['Item']['currentValue'])
next_id = curr + 1
counter_table.update_item(
Key={'counterId': 'counter'},
UpdateExpression='SET currentValue = :val',
ExpressionAttributeValues={':val': str(next_id)}
)
return f"CAM-{next_id:03d}"
頂いた指摘
・コメント不足
各処理の意図や想定ケースが読み手に伝わらず、保守しづらいという指摘。
・ 排他制御(トランザクション未対応)
DynamoDBとKinesis Video Streamsを連携しているが、どちらかの処理に失敗した際のロールバック処理が不十分。
・ ログ出力の粒度と出力先
ログの出力先・形式が不明瞭。ログレベル(INFO/ERROR等)の使い分けも不十分。
・ READMEの説明不足
APIの使用方法やログ仕様、使用しているPython標準関数の説明が未記載で、他の開発者が追従しづらい。
レビュー後のコード🍗
以下は、意図が読み取りやすくなった版です(参考コードから要点抜粋・改変)。
修正ポイント
・ログレベル付きのログ出力
・コメントの追加
・try-exceptの粒度を調整
・DynamoDBとKVSの操作を分離、戻り値もより明確化
import json
import boto3
import logging
# ロガー初期化
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# AWSリソース
dynamo = boto3.resource('dynamodb')
table = dynamo.Table('hogehogeDB')
counter_table = dynamo.Table('hogehogeCounter')
kvs = boto3.client('kinesisvideo')
def lambda_handler(event, context):
logger.info("Received event: %s", event)
# リクエスト解析
try:
body = json.loads(event.get('body', '{}'))
except Exception as e:
logger.error("JSON parse error: %s", e)
return response(400, "無効なJSON")
# 必須項目チェック
required = ['name', 'location', 'model', 'serialNumber']
missing = [k for k in required if not body.get(k)]
if missing:
return response(400, f"必須項目が不足しています: {', '.join(missing)}")
# カメラID採番
try:
id_resp = counter_table.get_item(Key={'counterId': 'counter'})
curr = int(id_resp.get('Item', {}).get('currentValue', 0))
next_id = curr + 1
counter_table.update_item(
Key={'counterId': 'counter'},
UpdateExpression='SET currentValue = :val',
ExpressionAttributeValues={':val': str(next_id)}
)
device_id = f"CAM-{next_id:03d}"
logger.info("採番されたDevice ID: %s", device_id)
except Exception as e:
logger.error("採番失敗: %s", e)
return response(500, "カメラIDの採番に失敗")
# KVSチャネル作成
try:
channel_name = f"webrtc-{device_id}"
kvs.create_signaling_channel(
ChannelName=channel_name,
ChannelType='SINGLE_MASTER',
SingleMasterConfiguration={'MessageTtlSeconds': 60}
)
logger.info("KVSチャネル作成: %s", channel_name)
except Exception as e:
logger.error("KVSチャネル作成エラー: %s", e)
return response(500, "KVSチャネル作成に失敗")
# DynamoDB登録
item = {k: body[k] for k in required}
item['deviceId'] = device_id
try:
table.put_item(Item=item)
logger.info("DynamoDB登録成功: %s", item)
except Exception as e:
logger.error("DynamoDB登録失敗: %s", e)
return response(500, "DynamoDB登録失敗")
return response(201, "登録成功", item)
def response(status_code, message, data=None):
return {
'statusCode': status_code,
'body': json.dumps({
'message': message,
'item': data if data else {}
})
}
感想
・心理的にコードレビューを始める前が一番抵抗強く感じました。
(正直動けばイイジャ...
ただ、一度ちゃんとやってみるとコードを書く際にどのような点を気を付けるべきなのか分かるようになりました。(視点の引き出しが増えました。感謝感謝)
・開発工数の見積もりも「レビュー対応込み」で精度が上がるようになりました。
(コードレビューの経験が無いと実装するのにはこれくらいかかるなというレビューまで考慮した見積もりは難しいなと感じます。)