S3に格納したCSVファイルのデータからCognitoへユーザ登録を行うLambdaプログラム

1.はじめに

内容はタイトルの通りですが、
S3に格納したCSVファイルを読み取り、
その情報でCognitoへユーザ登録を行うLambdaプログラムの説明です。
業務で製造する機会がありましたので自分の復習も兼ねて記載しようと思います。

2.前提

今回Lambda作成にはPythonを使用します。
また、Lambda作成前に以下は用意した状態で行っています。
・Cognitoユーザプール
・S3バケット

3.S3からCSVファイル読み込み

まずはS3に格納したCSVファイルの内容を読み取るところについてです。
S3には以下のような内容のCSVを格納しています。

user	mail
user1	[e-mail]
user2	[e-mail]

他にCognitoへ登録したい項目がある場合はその分の項目を追加することも可能です。

読み取りを行うプログラムは以下のようになります。

CSV読み込み

  #バケット名
  bucket_name = 'バケット名を入れる'
  #対象ファイル
  object_key = 'csvファイルのパスを入れる'

  # s3からcsvファイルを取得
  s3 = boto3.client('s3')
  csv_file = s3.get_object(Bucket=bucket_name, Key=object_key)
  csv_content = csv_file['Body'].read().decode('utf-8-sig')

  # CSVのデータを辞書形式で取得
  csv_data = csv.DictReader(csv_content.splitlines())

bucket_nameにはS3のバケット名を、
object_keyには使用するCSVファイル名を設定します。

boto3.client('s3')でS3へ接続するためのClientオブジェクトを取得できます。
そしてClientオブジェクトのget_objectメソッドに先ほど設定した
バケット名とCSVファイルパスを渡すことでCSVファイルオブジェクトを取得できます。

その後、csv_content = csv_file['Body'].read().decode('utf-8-sig')のところでは、
取得したCSVファイルオブジェクトのデータを取り出しています。

最後に、csv.DictReader(csv_content.splitlines())で
取り出したデータを辞書形式に変換して取得しています。
辞書形式で取得したデータは以下のような形で格納されており、
CSVファイルのヘッダーに指定した項目名を使用してデータを取り出せるようになります。

{'user': 'user1', 'mail': '[e-mail]'},
{'user': 'user2', 'mail': '[e-mail]'}

これでS3からCSVファイルのデータを取得することができましたので
次はCognitoへ取り出したデータを使用してユーザ登録をしていきます。

4.Cognitoへユーザ登録

以下がユーザ登録部分のプログラムになります。

ユーザ登録

  # Cognitoに接続するための情報を設定
  user_poolid = 'ユーザプールIDを入れる'
  cognito = boto3.client('cognito-idp', 'ap-northeast-1')

  # csvから取得したユーザ情報を行毎に取り出し、ユーザ登録処理を行う
  for row in csv_data:
    try:
      # ヘッダーにした項目名からパラメータを取得
      user_name = row['user']
      user_mail = row['mail']

      # ユーザ登録
      response = cognito.admin_create_user(
        # ユーザプールID
        UserPoolId=user_poolid,
        # ユーザ名
        Username=user_name,
        # メールアドレスの設定
        UserAttributes=[
          {
            'Name': 'email',
            'Value': user_mail
          },
        ],
      )
      
    except botocore.exceptions.ClientError as error:
      # ユーザ登録処理に失敗時の処理を記載

user_poolidには登録先のユーザプールIDを設定します。

まずはS3の時と同様に、boto3.client('cognito-idp', 'ap-northeast-1')で
Cognitoへ接続するためのClientオブジェクトを取得しています。

その後、for row in csv_data:で
S3から取得したデータを1行ずつ取り出し、登録処理を行っていきます。
データは辞書形式で取得しているため、user_name = row['user']のように
ヘッダーの項目名を指定して中身を取り出しています。

先ほど設定したユーザプールIDとS3から取り出した登録したいユーザデータを
cognito.admin_create_userで引数に設定することでCognitoへユーザ登録ができます。
今回はユーザ名とメールアドレスの登録のみですが、
他に登録したい項目がある場合は引数に追加することも可能です。
登録時にエラーが発生した場合の処理については、
except botocore.exceptions.ClientError as error:の下に記載できます。

これでユーザ登録もできました。

5.プログラム全体

ここまでの内容を含めたプログラム全体は以下になります。

Lambda全体

import csv
import boto3
import botocore

def lambda_handler(event, context):

  #バケット名
  bucket_name = 'バケット名を入れる'
  #対象ファイル
  object_key = 'csvファイルのパスを入れる'

  # s3からcsvファイルを取得
  s3 = boto3.client('s3')
  csv_file = s3.get_object(Bucket=bucket_name, Key=object_key)
  csv_content = csv_file['Body'].read().decode('utf-8-sig')

  # CSVのデータを辞書形式で取得
  csv_data = csv.DictReader(csv_content.splitlines())

  # Cognitoに接続するための情報を設定
  user_poolid = 'ユーザプールIDを入れる'
  cognito = boto3.client('cognito-idp', 'ap-northeast-1')

  # csvから取得したユーザ情報を行毎に取り出し、ユーザ登録処理を行う
  for row in csv_data:
    try:
      # ヘッダーにした項目名からパラメータを取得
      user_name = row['user']
      user_mail = row['mail']

      # ユーザ登録
      response = cognito.admin_create_user(
        # ユーザプールID
        UserPoolId=user_poolid,
        # ユーザ名
        Username=user_name,
        # メールアドレスの設定
        UserAttributes=[
          {
            'Name': 'email',
            'Value': user_mail
          },
        ],
      )
      
    except botocore.exceptions.ClientError as error:
      # ユーザ登録処理に失敗時の処理を記載

6.まとめ

"S3に格納したCSVファイルのデータからCognitoへユーザ登録を行うLambdaプログラム"
の説明は以上になります。

実装したコードについて、文字に起こす形で人に説明する機会はあまりないため、
記事を記載する中でより理解を深めることのできる、良い復習になりました。

7.参考