AWS Lambda에서 RDS(PostgreSQL) 연동하기

읽기 전

• 불필요한 코드나 잘못 작성된 내용에 대한 지적은 언제나 환영합니다.
• 개인적으로 사용해보면서 배운 점을 정리한 글입니다.

이전 포스팅에서 psycopg2 의존성 문제 때문에 pip 대신 poetry를 채택 후 사용법에 대해 정리하였다. 다른 이가 미리 빌드한 파일도 찾으면 금방 나오지만 아무래도 module의 버전 관리에 취약하다는 문제가 있어 직접 해봤다. 이번 포스팅에서는 psycopg2 module을 빌드하여 layer에 업로드 후 RDS에 query request하여 response를 수신하는 과정까지 수행하는 내부 모듈을 작성해보려 한다. 원래 별도의 lambda 함수로 만들었는데 비용 측면에서 옳지 못하다는 생각이 들어 모듈화를 진행하였다.

AWS RDS 생성 및 외부 연결

참고 링크 - AWS RDS DB 생성 및 외부 연결하기

Poetry로 Python module 다루기

참고 링크 - Poetry로 python module 관리하기

psycopg2 module 파일 구성

psycopg2 의존성 문제

첨언을 하자면 psycopg2는 PostgreSQL에 접근하여 데이터를 다룰 수 있게 만드는 libpq 라이브러리를 python에서 사용할 수 있게 해주는 python module이다. 그러니 libpq라는 C언어 기반 라이브러리를 python 환경에서 동작할 수 있게끔 하는 wrapper이기 때문에 단순 설치로는 당연히 에러가 발생한다. 때문에 poetry add psycopg2 하기 전 psycopg2가 의존하고 있는 libpq-dev 라이브러리 설치가 선행되어야 정상적으로 실행된다. 그러나 어떤 module 의존성을 갖는지 일일히 체크해서 설치하기엔 나같이 인내심이 부족한 사람들을 위해 정식으로 psycopg2-binary 모듈을 함께 관리해주고 있다.

psycopg2 module 설치

pip install psycopg2 -t [폴더경로]를 실행해서 설치한 뒤 layer에 업로드하면 import에러가 발생한다. 원인은 Amazon Machine Image에 psycopg2가 의존하는 libpq가 없기 때문에 같이 올려줘야 한다. 이전 포스팅인 Poetry로 python module 관리하기에서 poetry를 통해 psycopg2-binary 모듈을 설치하자. 일반적인 module 설치를 진행하는 다음 명령어는 psycopg2 설치 시 오류를 발생시킨다.
poetry add psycopg2

오류를 해결하기 위해 stand-alone 버전으로 만들어진 psycopg2-binary를 다음 명령어를 실행하여 설치하자.

poetry add psycopg2-binary

그 후 layer에 업로드하기 위해선 module 파일을 설치해야 한다. pip로 설치하기 전 poetry가 관리하던 module과 의존성 정보를 다음 명령어를 실행하여 export하자.
poetry export -f requirements.txt --output requirements.txt
requirements.txt를 추출하였으면 해당 파일을 적당한 위치에 넣고 wsl이던 window cmd이던 requirements.txt를 기반으로 module을 설치하는 다음 명령어를 실행한다.
pip intsall -r [requirements.txt 경로] -t [module 설치 경로]
그럼 다음과 같이 layer에 업로드할 파일이 구성된다.

psycopg2 module로 RDS(PostgreSQL) 연결하기

module 파일을 업로드하여 layer를 생성하거나 기존 layer에 새로운 버전을 publish 했다면 사용하고자 하는 lambda에 layer를 연결하여 psycopg2 module을 import한다. RDS에 연결하기 위해서는 Database Instance 정보가 필요한데 코드 상에 static하게 관리하면 보안 상의 문제가 발생할 여지가 있으므로 lambda에서 제공하는 환경변수의 형태로 관리한다. 다음 그림은 환경 변수 등록하는 UI와 DB Instance에 연결하는 코드다.

코드 실행 시 에러 없이 정상적으로 동작함을 확인했으니 query request를 받고 response를 반환할 내부 모듈을 작성해보자.

VPC 및 IAM 권한 설정하기

다른 인스턴스를 호출하기 위해선 서로 같은 VPC 안에 있어야 한다. Lambda 콘솔에서 VPC 설정을 해주자. 그리고 IAM에서 Lambda 함수 역할에 VPC를 Execute할 수 있는 권한도 추가해야 한다.

RDS 호출 및 query 실행 코드 작성

일반적으로 CRUD 작업을 수행하니 필요한 query 명령어는 SELECT, UPDATE, INSERT, UPDATE 정도로 생각한다. 다음 코드는 CRUD에 필요한 쿼리를 실행하는 기능을 갖는다. 디렉토리 구조는 다음과 같다.

call-rds-postgreSQL  - lambda 함수 이름
├── db - db 내부 module 폴더
|    └── db.py - psycopg2 db query handler 함수
└── lambda_function.py - 메인 함수

db/db.py 코드

import inspect
import psycopg2
import os
import re

def execute(query):
    with psycopg2.connect(host=os.environ["DB_HOST"], user=os.environ["DB_USER"],
    dbname=os.environ["DB_NAME"], password=os.environ["DB_PASSWORD"]) as conn:
        with conn.cursor() as cur:
            try:
                cur.execute(query) # 쿼리 실행
                # _로 구분된 예약어 중복 제거
                cols = [des.name.rstrip('_') for des in cur.description]
                rows = cur.fetchall() # select 쿼리 실행 시 결과 조회
                # col=[a,b,c], row=[(1,2,3),(4,5,6)]의 형태를 [{a:1, b:2, c:3}, ...]로 매핑
                res = [dict(zip(cols, row)) for row in rows]
                return res
            except Exception as e:
                print(str(e))
                cur.close()

def sanitize(func):
    """
    db 인자를 다루는 함수에서 사용해야 하는 decorator
    Args:
        func: db 인자를 다루는 함수
         - func함수는 dict 형태를 인자로 받아야 함
    Returns: SQL escaping된 인자들로 치환된 func
    """
    def wrapper(*args, **kwargs):
        # default 명시된 인자 목록 확인
        default_params = (
            (k, v.default)
            for k, v in inspect.signature(func).parameters.items()
            if v.default is not inspect.Parameter.empty
        )
        # default 명시된 인자에 대해 값을 받지 못하면 default값 입력
        for k, v in default_params:
            if k not in kwargs:
                kwargs[k] = v
        """
        python의 None을 'NULL'로 치환
        python방식의 quoting 포맷을 SQL 방식으로 치환
        SQL Injection 공격 대비 특수문자 제거
        이후 새로운 인자가 반영된 함수 반환
        """
        for k, v in kwargs.items():
            if v is None:
                kwargs[k] = 'NULL'
            elif isinstance(v, str):
                v = re.sub('[\s\t\'\/\\\;=&+!#$%*"]', '', v)
                v = v.replace("'", "''")
                kwargs[k] = "'" + v + "'"
        return func(*args, **kwargs)
    return wrapper

lambda_function.py 코드

import json
from db import db

@db.sanitize
def insert(grade, class_code, subject_code, name=None, score=None):
    return db.execute(f"""INSERT INTO practice.grade_test (grade, class_code, subject_code, name, score) VALUES
        ({grade}, {class_code}, {subject_code}, {name}, {score}) RETURNING *;""")

@db.sanitize
def select():
    return db.execute(f"SELECT * FROM practice.grade_test;")

@db.sanitize
def delete(name=None):
    return db.execute(f"DELETE FROM practice.grade_test WHERE name={name} RETURNING *;")

@db.sanitize
def update(grade, class_code, subject_code, name, score):
    return db.execute(f"""UPDATE practice.grade_test SET score={score} 
    WHERE grade={grade} AND class_code={class_code} AND subject_code={subject_code} AND name={name}
    RETURNING *;""")

def lambda_handler(event, context):
    """"
    input = dict()
    input['grade'] = 1
    input['class_code'] = 1
    input['subject_code'] = 1
    input['name'] = "ABC가나12"
    input['score'] = 12345
    return insert(**input)
    """
    #return select()
    """
    input = dict()
    input['grade'] = 1
    input['class_code'] = 1
    input['subject_code'] = 1
    input['name'] = "ABC가나12"
    input['score'] = 54321
    return update(**input)
    """
    """
    input = dict()
    input['name'] = "ABC가나12"
    return delete(**input)
    """

메인 함수에서 쿼리를 요청할 내부 module 함수를 정의했으니 적당히 메인함수에서 CRUD 쿼리를 생성하여 테스트를 하면 잘 작동함을 확인할 수 있다. 다음 포스팅에서는 lambda - graphql - RDS를 연동하여 API 작성에 대해 정리하려 한다.