LoginSignup
0
0

More than 1 year has passed since last update.

@zumax

Amazon CodeWhispererでLambdaのコード（Python）を実装してみた

Last updated at Posted at 2023-04-22

背景・目的

先日、AWS Glue データカタログへのクロスアカウントアクセスを試してみたでは、Athenaから別アカウントのData Catalogにアクセスしましたがこれを、Lambdaを使用して取得したいと思います。

このLambdaを、先日、一般提供が発表された「Amazon CodeWhisperer」を使用して、実装してみようと思います。

image.png

まとめ

  • CodeWhisperでは、コードの補完、コメントからコードを生成できます。
  • LambdaコンソールとIDEでは、下記の違いがあります。
    • Lambdaコンソールの場合は、単一のSuggest、手動で取得
    • IDEの場合は、複数のSuggest、自動で取得されるようです。

概要

What is CodeWhisperer?

  • リアルタイムでコードの推奨事項を提供する、機械学習を利用した汎用コードジェネレーター
  • コードを書くと、CodeWhisperer は既存のコードとコメントに基づいて自動的に提案を生成する。
  • パーソナライズされたレコメンデーションは、1 行のコメントから完全に形成された関数まで、サイズと範囲がさまざま。
  • CodeWhisperer はコードをスキャンして、セキュリティの問題を強調して定義することもできる。
  • CodeWhisperer がサポートするプログラミング言語については、言語サポートを参照。

Amazon CodeWhisperer Features

Support for popular programming languages and IDEs

一般的なプログラミング言語と IDE をサポートしています。
下記の言語に対して、コードSuggestを提供

  • Python
  • Java
  • JavaScript
  • TypeScript
  • C#
  • Go
  • Rust
  • PHP
  • Ruby
  • Kotlin
  • C、C++
  • シェル スクリプト
  • SQL
  • Scala

下記のIDEで利用できます。

  • JetBrains IDE (IntelliJ IDEA、PyCharm、WebStorm、Rider)
  • Visual Studio (VS) Code
  • AWS Cloud9
  • AWS Lambda コンソールなど

Built-in security scans

  • 上位 10 の Open Worldwide Application Security Project (OWASP) に含まれる脆弱性や、暗号ライブラリのベスト プラクティスを満たさない脆弱性など、見つけにくい脆弱性を検出できる。

Enterprise administration

SSO と AWS IAM Identity Center の統合を使用して、ユーザーまたはグループに CodeWhisperer へのアクセスを提供する。

Amazon CodeWhisperer Pricing

下記の2つのTierがあります。

  • CodeWhisperer Individual Tier
    • 無料
    • 一ヶ月あたり最大50回のコードスキャン/ユーザが無料で利用できる
  • CodeWhisperer Professional Tier
    • 「ユーザーごと、月ごと」に計算され、暦月の請求期間中に CodeWhisperer にアクセスできるユーザーの最大数に基づいて、組織に毎月請求される。

Security scans

  • セキュリティ スキャンの実行は、IDE エディタで現在アクティブなファイルと、プロジェクトの依存ファイルに対してセキュリティ スキャンを実行。 スキャンが終了すると、スキャンされたファイルのセキュリティの問題が VSC の問題パネルで強調表示される。
  • CodeWhisperer のセキュリティ スキャンは、Amazon CodeGuru と統合されている。

実装

バージニアリージョンで実行しています。

事前準備

Lambda関数を作成

まずは、Lambda関数を作成します。ここではcross-account-by-lambdaという関数名で作成しました。
image.png

IAMポリシーの設定

  1. 上記のLambda関数作成時に、自動でIAMロールを作成しました。このIAMロールのポリシーに下記を追加します。

    {
    "Sid": "CodeWhispererPermissions",
    "Effect": "Allow",
    "Action": [
        "codewhisperer:GenerateRecommendations"
    ],
    "Resource": "*"
}

  2. GlueカタログにアクセスするGlueのIAMポリシーを追加します。

    {
    "Effect": "Allow",
    "Action": [
        "glue:CreateDatabase",
        "glue:DeleteDatabase",
        "glue:GetDatabase",
        "glue:GetDatabases",
        "glue:UpdateDatabase",
        "glue:CreateTable",
        "glue:DeleteTable",
        "glue:BatchDeleteTable",
        "glue:UpdateTable",
        "glue:GetTable",
        "glue:GetTables",
        "glue:BatchCreatePartition",
        "glue:CreatePartition",
        "glue:DeletePartition",
        "glue:BatchDeletePartition",
        "glue:UpdatePartition",
        "glue:GetPartition",
        "glue:GetPartitions",
        "glue:BatchGetPartition"
    ],
    "Resource": [
        "*"
    ]
}

  3. Athenaの出力結果を格納するS3バケットのIAMポリシーを追加します。

    {
    "Effect": "Allow",
    "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:ListBucket"
    ],
    "Resource":[
        "arn:aws:s3:::{利用者のAthenaの結果を格納するバケット}",
        "arn:aws:s3:::{利用者のAthenaの結果を格納するバケット}/*"
    ]
}

  4. 提供側のS3バケットから取得する権限を付与します。

    {
    "Effect": "Allow",
    "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:ListBucket"
    ],
    "Resource":[
        "arn:aws:s3:::{提供側のデータを格納するバケット}",
        "arn:aws:s3:::{提供側のデータを格納するバケット}/*"
    ]
}

  5. 後ほどAthenaを実行するので、IAMポリシーをアタッチします。
    image.png

CodeWhisperを有効化

  1. Lambdaのコードソースで、Tools>「Amazon CodeWhisperer Code Suggestions」を選択し、有効にします。
    image.png

  2. Preferencesで「Amazon CodeWhisperer Code Suggestions」が有効になっていることを確認します。
    image.png

その他の設定

  1. この後、Athenaを使用してクエリを実行します。デフォルトの３秒だと短いので、タイムアウト時間を3分に延ばしておきます。
    image.png

デモ

Introducing Amazon CodeWhisperer in the AWS Lambda console (In preview)を参考に、操作感を試してみます。

Lambdaコンソールの場合は、単一のSuggest、手動で取得、IDEの場合は、複数のSuggest、自動で取得されるようです。

以降から、Macを前提として記載しています。

  • Suggestを使うには、ショートカットキーを使用します。
    • Suggestを手動で取得する
      • Macの場合、Option+C
      • Windowsの場合、Alt+C
    • Suggestを受け入れる場合
      • Tab
    • Suggestを拒否
      • ESCまたは、Backspace

とりあえず試す

  1. import json直下でOption+C
    image.png

  2. Tabで決定
    image.png

Single-line code completion

行単位でコードの補完を行ってくれるか確認します。
:::note info
変数名によりSuggestが変わりました。
:::

  1. athenaと途中まで書いてOption＋CとTabを実行。boto3の初期化をSuggestしてくれました。
    image.png

    image.png

  2. 次にs3と書いてみました。
    image.png
    image.png

Full function generation

関数単位で、コードの生成を行ってくれるか確認します。

  1. def read_from_s3と書いて、Option+CとTabを実行。テンプレートを書いてくれました。
    image.png
    image.png

Generate code from comments

コメントからコードの生成を行ってくれるか確認します。

  1. Function to upload a file to an S3 bucketとコメントに書いて、Option+C＋Tabを実行。
    image.png

  2. そのまま連続でOptopn+Cを実行すると続けてSuggestしてくれました。
    image.png

Athenaのクライアントプログラムを書いてみる

2つの関数を作成する

Athenaに対してクエリを実行する関数と、クエリが正常に終了したか確認する関数を実装します。

  1. 初期状態は下記の通りです。
    image.png

  2. Function to query using Athenaとコメントを入力します。query_athenaという関数が生成されました。
    image.png

  3. そのままOptions+Cを打ち込みます。雛形ができました。
    image.png
    image.png

  4. 次に、Athenaはクエリの実行結果をチェックする必要があるので、Function to check the result of "query_athena" function (input is query ID, output is true or false)と入力します。check_query_statusという関数が生成されました。

  5. そのままOptions+Cをクリックします。コードが生成されました。
    image.png
    image.png

仕様変更（戻り値の変更）

上記で作成したquery_athena関数の戻り値がresponseだったのでクエリIDを戻したいと思います。

  1. # return to queryIdとコメントし、Option+C & Tabで追加されました。
    image.png

  2. 元のコード（return response）は削除しておきます。
    image.png

仕様変更（仕様の変更）

Athenaのクエリ実行とチェックは非同期のため、実行後すぐに結果をチェックしても終了していない可能性があります。そのため最大値を設けて、定期的に確認するように仕様を変更します。（20秒ごとに最大3回確認します。）

  1. コメントを追加し、Options+C＆Tabを実行します。生成されました。
    image.png
    image.png

作成した関数を呼び出す

  1. 関数（query_athenaとcheck_result）を呼び出すコードを生成します。
    image.png

残りのクエリを修正します。

下記については、手動で直します。

  1. Athenaの出力結果であるS3バケット名を指定します。

  2. 接続先を

  3. クエリを変更します

    query_id = query_athena("SELECT * FROM account_a_catalog.crossaccount_access_db.test_table")

  4. check_result関数のロジックが、このままでは一回目のチェックでいきなり抜けてしまうので、SUCCEEDEDではないときにSleepするように修正します。

    # Function to check the result of "query_athena" function (input :query ID, output :true or false) . 
# Check the result of "query_athena" function every 20 seconds for max 3 times.
def check_result(queryId):
    state = "RUNNING"
    client = boto3.client('athena')
    max_execution = 3
    
    while max_execution > 0 and state in ["RUNNING", "QUEUED"]:
        max_execution -= 1
        response = client.get_query_execution(QueryExecutionId=queryId)
        if "QueryExecution" in response and "Status" in response["QueryExecution"] and "State" in response["QueryExecution"]["Status"]:
            state = response["QueryExecution"]["Status"]["State"]
            if state == 'SUCCEEDED':
                return True

        time.sleep(20)

    return False

  5. sleepを使用するためにimport timeを追加します。

動作確認

  1. Lambdaのコンソールから「テスト」をクリックします。
    image.png

  2. Athenaの結果を確認します。ステータスが「完了済」になっています。
    image.png

  3. 結果をダウンロードします。
    image.png

  4. 確認できました。
    image.png

考察

今回、CodeWhisperer を使用してLambdaをコード（Python）を実装してみました。
今までは、boto3のAPIなど、マニュアルを読んでコーディングするところから開始していましたが、CodeWhispererをすることで、ある程度実装された状態から始められる点は良いと感じました。

なお、今回はLambdaのコンソールを使用しましたが、IDEは複数のSuggestとキー入力が不要ということなので、今後はIDEで試してみたいと思います。

参考

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0