API Gateway + Lambda + RDS로 서버리스 API 만들기

Lambda가 VPC 내의 리소스에 액세스 할 수 있도록 AWSLambdaVPCAccessExecutionRole 권한을 추가해준다.

RDS에 접근하기 위해 같은 VPC를 연결해준다.

프라이빗 서브넷은 NAT 게이트웨이를 통해 인터넷에 액세스할 수 있지만, 함수를 퍼블릭 서브넷에 연결해도 인터넷 액세스가 제공되지는 않는다.
그러므로 secretsmanager 엔드포인트를 생성해준다.

Lambda Code

import boto3
import pymysql
import json

def get_secret():
    client = boto3.client('secretsmanager', region_name='ap-northeast-2')
    secret = client.get_secret_value(
        SecretId='lambda-sm'
    )
    return json.loads(secret['SecretString'])

def get_conn():
    secret = get_secret()
    try:
        conn=pymysql.connect(
            host=secret['host'],
            user=secret['username'],
            password=secret['password'],
            port=secret['port'],
            db=secret['dbname']
        )
        print("DB 연결 성공")
        return conn
    except Exception as e:
        print(f"DB 연결 실패: {e}")

def resp(status_code, body):
    return {
        'statusCode': status_code,
        'body': json.dumps(body, ensure_ascii=False, default=str)
    }

def get_employees():
    conn = get_conn()
    cur = conn.cursor(pymysql.cursors.DictCursor)
    sql  = "select * from employees order by id asc"
    cur.execute(sql)
    ids = cur.fetchall()
    return resp(200, {'employees': ids, 'count': len(ids)})

def get_employee(employee_id):
    conn = get_conn()
    cur = conn.cursor(pymysql.cursors.DictCursor)
    sql = "select * from employees where id = %s"
    cur.execute(sql, (employee_id,))
    id = cur.fetchone()
    if not id:
        return resp(404, {'error': f'Employee {employee_id} not found'})
    return resp(200, {'employee': id})

def create_employee(data):
    try:
        conn = get_conn()
        cur = conn.cursor(pymysql.cursors.DictCursor)
        sql = "insert into employees (name, department, position, email, hire_date, phone, salary) values (%s, %s, %s, %s, %s, %s, %s)"
        cur.execute(sql, (data['name'], data['department'], data['position'], data['email'], data['hire_date'], data['phone'], data['salary']))
        conn.commit()
        # lastrowid는 DB에 INSERT 하고 나서 방금 생성된 row의 ID를 가져오는 것
        return resp(201, {'msg': "Employee created", "id": cur.lastrowid})
    except pymysql.err.IntegrityError:
        return resp(409,{'error': f"Email {data['email']} already exists"})
    except Exception as e:
        return resp(400, {'msg': f'Missing field: {e}'})

def delete_employee(employee_id):
    conn = get_conn()
    cur = conn.cursor(pymysql.cursors.DictCursor)
    sql = "delete from employees where id = %s"
    cur.execute(sql, (employee_id,))
    conn.commit()
    # 바뀐게 0개이면 404
    if cur.rowcount == 0:
        return resp(404, f"Employee {employee_id} not found")
    return resp(200, { 'msg': 'Deleted', 'id': employee_id })

def lambda_handler(event, context):
    method = event.get('httpMethod', '')
    path = event.get('path', '')
    params = event.get('pathParameters') or {}
    employee_id = params.get('id')
    body = json.loads(event['body']) if event.get('body') else {}

    if method == 'GET'    and path == '/employees': return get_employees()
    if method == 'GET'    and employee_id:          return get_employee(int(employee_id))
    if method == 'POST'   and path == '/employees': return create_employee(body)
    if method == 'PUT'    and employee_id:          return update_employee(int(employee_id), body)
    if method == 'DELETE' and employee_id:          return delete_employee(int(employee_id))

    return resp(405, {'error': f'Method not allowed: {method} {path}'})

API Gateway에서 메서드를 생성해준다.
Lambda Proxy 통합을 해줘야 lambda에서 statusCode/headers/body를 직접 제어할 수 있다.