#18 AWS 람다 - 스크래핑한 데이터를 DB에 저장하기

람다 코드를 먼저 작성했고, 이땐 타입힌트나 sqlalchemy에 대해 미숙했기 때문에 뭔가 이상한 점이 있을 수 있습니다.
다른 기능을 만드느라 큰 문제가 발생하지 않으면 수정을 하지 못하는 점 양해 부탁드립니다!

 

AWS 람다는 크롤러 / 스크래퍼를 구현하기에 아주 좋다.
서버리스로 애플리케이션을 띄울 수 있고, python / node.js로 구현도 부담스럽지 않다. (종속성 모듈도 사용할 수 있다)
AWS서비스에서 이벤트를 전송하는 기능도 있고, 로그를 모니터링 해주기도 한다.
또, EventBridge Scheduler를 이용하면 주기적으로 람다를 실행시킬 수도 있다.

하지만 람다의 보안상 제약이 있다. 
VPC안에서는 외부 API를 호출할 수 없다는 제약이다.
VPC밖에서는 반대로 RDS에 접근할 수 없다.

그래서 좋은 방법이 없을까 해서 아래의 글에서 힌트를 얻었다.
https://serverlessfirst.com/lambda-vpc-internet-access-no-nat-gateway/

키 포인트는 같은 VPC에 속해있지 않더라도 람다끼리는 호출이 가능하다는 것이다.
이를 이용해서 아래의 그림 처럼 람다를 두개로 나눠볼 수 있다.

전체적으로 스크래퍼는 스팀API와 RDS에 여러번 접근해야하기 때문에 
유동적으로 두 람다가 상효작용 할 수 있어야 한다.
그래서 람다를 호출할때 같이 보낼 수 있는 페이로드와 / 람다가 반환하는 결과를 이용하기로 했다. 
이 페이로드와 결과는 json형태로 되어있으며, 파이썬에서는 딕셔너리 형태로 사용한다.

페이로드 딕셔너리는 Event라고 추상화했다.
이 Event(이벤트) 행동을 나타내는 name(이름)과 값을 나타내는 payload로 구성되어있다.

# event.py
EventName = Literal[
    "save_games",
    "save_screenshots",
    "get_games_in_steam_ids",
    "get_some_games",
    "get_screenshots_in_steam_file_ids",
]

class Event(TypedDict):
    name: EventName
    payload: Any

public lambda에서는 lambda client를 wrapping한 LambdaAPI 클래스를 만들었다.
이 객체를 이용해 private lambda를 호출하는데, 항상 이벤트 객체 단위로 호출하도록 했다.

class LambdaAPI(protocols.LambdaAPI):
    def __init__(self, private_function_name: str) -> None:
        self.private_function_name = private_function_name
        self.client = boto3.client("lambda")

    def invoke_lambda(self, event: Event) -> Any:
        response = self.client.invoke(FunctionName=self.private_function_name, Payload=json.dumps(event))

        if "FunctionError" in response:
            raise AWSLambdaException(response["FunctionError"])

        payload: Any = response["Payload"].read().decode("utf-8")
        return json.loads(payload)
    ….

호출된 private lambda는 실행되면서 진입점인 lambda_handler에서 시작된다.
이 곳에서는 이벤트를 받고 handle_event()메서드를 통해 그 맞는 동작을 한다.

# lambda_func.py
def lambda_handler(event: Event, context: Any):
    engine = create_engine(config.DATABASE_URL)
    init_database(config, engine)

    logger.info("Handle event [required event is %s]", event)

    result = handle_event(engine, event)

    logger.info("Result: %s", result)

    return result

 


handle_event 메서드는 미리 이벤트에 매핑해둔 함수들을 찾아 페이로드를 인자로 넣고 실행한다.
함수의 반환값은 항상 json으로 변환될 수 있는 형태로 반환되며 이 값을 public lambda로 전송한다.

# lambda_func.py
funcs: dict[EventName, Callable[..., Any]] = {
    "save_games": save_games,
    "get_some_games": get_some_games,
    "get_games_in_steam_ids": get_games_in_steam_ids,
    "save_screenshots": save_screenshots,
    "get_screenshots_in_steam_file_ids": get_screenshots_in_steam_file_ids,
}


def handle_event(engine: Engine, event: Event) -> Any:
    with Session(engine) as session:
        func = funcs[event["name"]]

        try:
            result = func(session, event["payload"])
        except TypeError:
            result = func(session)

        session.commit()
        return result