지금의 시대, 자신이 보유하고 있는 디지털 데이터의 백업은 필수가 되고 있습니다. Qiita의 투고도 언제 망가질지 알았던 것이 아닙니다. 그래서 AWS Lambda에서 Qiita API를 두드려 백업을 수행하는 메커니즘을 만듭니다.

전제

Qiita API V2 사용

AWS Lambda에 Python3에서 스크립트 작성

백업 대상은 S3입니다

정기 실행을 가정하여 중복 데이터는 다시 백업하지 않습니다 (조금 지능적으로)

아티팩트

게시물 검색 처리

Qiita API의 페이지에서 개설되고 있는 바와 같이, 이하와 같은 URL로 인증 없이 투고 내용은 취득을 할 수 있으므로, 이번은 이것을 이용합니다.

https://qiita.com/api/v2/users/{user_id}/items?page={page}&per_page=100

처리 중 100건 이상을 취득할 수 있도록 루프를 합니다. 다음 페이지 수가 0개가 될 때까지 반복합니다.

    while is_loop:
        url = QIITA_BASE_URL.format(user_id = name, page = page)
        result = requests.get(url)
        result_json = json.loads(result.text)
        response_json.extend(result_json)
        page+=1
        if len(result_json) == 0:
            is_loop = False

취득한 처리의 내용을 확인해, URL(식별자로 한다)를 추출해, 배열로 합니다. 나중에 마찬가지로 S3에 저장된 저장된 URL 목록(JSON 배열)도 여기에서 URL 전용 배열로 변환합니다.

def extract_keys_as_array(target_json_array, key):
    result_array = []
    for target in target_json_array:
        result_array.append(target[key])

    return result_array

에서 이미 S3에 저장된 (두 번째 이후) JSON 데이터와 비교하여 과부족이 있는지 확인합니다. 과부족이 없으면 아무것도 하지 않도록 합니다. S3로부터의 취득 처리는 이런 느낌이 됩니다 s.

def get_topics():
    json_file = SETTING_FILE_NAME + ".json"
    response_data = []

    try:
        response = s3_client.get_object(Bucket=S3_BUCKET, Key=json_file)
        response_data = json.loads(response['Body'].read())
    except botocore.exceptions.ClientError as ex:
        if ex.response['Error']['Code'] == 'NoSuchKey':
            print("No Objects.")

    return response_data

저장된 URL 목록이 현재 주제와 함께 있는지 여부는 다음과 같습니다.

def is_same_topics(org_topics_array, target_array):
    is_same = True

    for org in org_topics_array:
        if org in target_array:
            target_array.remove(org)
        else:
            is_same = False
            break

    return is_same

과부족이 있으면 새로운 주제가 있다고 가정하고 JSON을 S3에 저장하십시오. 그런 다음 저장된 URL 목록도 업데이트합니다.

def put_data_to_s3(contents, file_name):

    tmp_dir = "/tmp/"
    tmp_file = file_name + ".json"

    with open(tmp_dir + tmp_file, 'w') as file:
        file.write(json.dumps(contents, ensure_ascii=False, indent=4, sort_keys=True, separators=(',', ': ')))

    s3_client.upload_file(tmp_dir + tmp_file, S3_BUCKET, tmp_file)

    return True

호출자는 이런 느낌입니다.

def run(event, context):
    result = "No need to update."
    response_json = get_qiita_post(event['name'])
    org_topics = get_topics()

    if len(org_topics) > 0:
        target_array = extract_keys_as_array(response_json, 'url')
        org_array = extract_keys_as_array(org_topics, 'url')

        if not is_same_topics(org_array, target_array):
            put_data_to_s3(response_json, BACKUP_FILE_NAME)
            put_data_to_s3(extract_keys_as_json(response_json, 'url'), SETTING_FILE_NAME)
            result = "Update finished."

    return result

실행해보기

첫 번째는 저장되었지만 두 번째는 이미 있는 URL 목록과 S3에 저장된 URL 목록이 동일하기 때문에 저장이 수행되지 않았습니다.

$ sls invoke local -f run -d '{"name": "kojiisd"}'
"Update finished."
$ sls invoke local -f run -d '{"name": "kojiisd"}'
"No need to update."

이런 느낌으로 S3에게는 무사히 보존되었습니다. (파일명은 환경 변수로 지정할 수 있게 되어 있습니다)

요약

일단 필요한 정보를 저장할 수 있게 되었습니다. 이번에는 S3를 보존처로 했습니다만, 인터페이스를 변경하는 것으로 DynamoDB등의 다른 스토리지에의 보존도 간단하게 실현할 수 있을 것 같습니다. 나중에 이 스크립트를 정기 실행하도록 Lambda 스케줄러에서도 설정하면 언제든지 Qiita의 백업은 취할 수 있게 됩니다.