前回、Amazon Bedrockを試した流れで、Excelの設計書からAWSのAPIを作成する検証をしてみました。
以下の記事の続きになっているので、詳細は以下を参照してください。
Cloud9を前提としています。
今回の内容を試すには、以下のパッケージをインストールする必要があります。
$ pip install boto3
$ pip install langchain
$ pip install unstructured
$ pip install networkx
$ pip install pandas
$ pip install openpyxl
Excelの設計書を読み込んでコードを作成するPythonスクリプトです。
test-document-loader
from langchain.document_loaders import UnstructuredExcelLoader
from langchain.prompts import PromptTemplate
from langchain.llms import Bedrock
prompt = '''
Human: あなたはAWSのエキスパートです。指定された設計書のデータに基づき、
コードを作成してください。コードの種類は、Lambdaのコード、テストコード、
AWS SAMで必要なtemplate.yamlです。
<document>
{document}
</document>
Assistant:
'''
inference_modifier = {'max_tokens_to_sample':4096,
"temperature":0.5,
"top_k":250,
"top_p":1,
"stop_sequences": ["\n\nHuman"]
}
llm = Bedrock(
region_name="us-east-1",
model_id="anthropic.claude-v2:1",
model_kwargs = inference_modifier
)
loader = UnstructuredExcelLoader("設計書.xlsx", mode="elements")
docs = loader.load()
data = [str(doc.page_content) for doc in docs]
print(llm(prompt.format(document="\n".join(data))))
使用したExcelの設計書は以下の内容です。
オブジェクトは認識できなかったので、テキストのみです。
数分で作成した雑な内容となっています。
1. API Gateway,
種類,REST
リソースパス,/show-item
HTTPメソッド,POST
,
2. Lambda,
関数名,ShowItemFunction
ランタイム,Python3.9
メモリ,128MB
タイムアウト,30秒
実行ロール,DynamoDBへのアクセス権限
,
3. DynamoDB,
テーブル名,Test
プライマリーキー,id
属性,item
属性型,文字列
,
4.処理内容,
1,idをクエリパラメータとして受け取る
2,idパラメータが存在しない場合はエラーで終了
3,Testテーブルからidをキーとしてレコードを取得
4,レコードが存在しない場合はエラーで終了
5,idとitemをJSONデータとしてbodyに格納し正常終了(200OK)
上記Excelデータを読み込んでコードを生成してもらいました。
この回答をもとにファイルに直しました。
はい、指定された要件に基づきコードを作成します。
Lambda関数のコードは以下のようになります。
```python
import json
import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Test')
def lambda_handler(event, context):
# idを取得
id = event['queryStringParameters']['id']
if id is None:
return {
'statusCode': 400,
'body': json.dumps({'message': 'id is required'})
}
# DynamoDBからデータを取得
resp = table.get_item(Key={'id': id})
if 'Item' not in resp:
return {
'statusCode': 404,
'body': json.dumps({'message': 'item not found'})
}
# レスポンスを作成
item = resp['Item']['item']
return {
'statusCode': 200,
'body': json.dumps({'id': id, 'item': item})
}
```
テストコードは以下のようになります。
```python
import unittest
from show_item import lambda_handler
class TestShowItem(unittest.TestCase):
def test_success(self):
event = {'queryStringParameters': {'id': 'test_id'}}
resp = lambda_handler(event, None)
self.assertEqual(resp['statusCode'], 200)
body = resp['body']
data = json.loads(body)
self.assertEqual(data['id'], 'test_id')
self.assertEqual(data['item'], 'test_item')
def test_missing_id(self):
event = {}
resp = lambda_handler(event, None)
self.assertEqual(resp['statusCode'], 400)
def test_item_not_found(self):
event = {'queryStringParameters': {'id': 'invalid_id'}}
resp = lambda_handler(event, None)
self.assertEqual(resp['statusCode'], 404)
if __name__ == '__main__':
unittest.main()
```
SAMテンプレートは以下のようになります。
```yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Sample SAM Template
Resources:
ShowItemFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: show_item/
Handler: app.lambda_handler
Runtime: python3.9
MemorySize: 128
Timeout: 30
Policies:
- DynamoDBCrudPolicy:
TableName: Test
Events:
GetResource:
Type: Api
Properties:
Path: /show-item
Method: post
TestTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: Test
AttributeDefinitions:
- AttributeName: id
AttributeType: S
KeySchema:
- AttributeName: id
KeyType: HASH
ProvisionedThroughput:
ReadCapacityUnits: 1
WriteCapacityUnits: 1
```
以上です。ご確認のほどよろしくお願いします。
Lambdaのコードです。
以下の2点を修正しました。
・boto3.resourceにregion_nameを追加(Cloud9でテストを通すため)
・queryStringParametersの存在有無をチェック(テストコードが通らなかったため)
app.py
import json
import boto3
#dynamodb = boto3.resource('dynamodb')
dynamodb = boto3.resource('dynamodb', region_name="ap-northeast-1")
table = dynamodb.Table('Test')
def lambda_handler(event, context):
if not 'queryStringParameters' in event:
return {
'statusCode': 400,
'body': json.dumps({'message': 'queryStringParameters is required'})
}
# idを取得
id = event['queryStringParameters']['id']
if id is None:
return {
'statusCode': 400,
'body': json.dumps({'message': 'id is required'})
}
# DynamoDBからデータを取得
resp = table.get_item(Key={'id': id})
if 'Item' not in resp:
return {
'statusCode': 404,
'body': json.dumps({'message': 'item not found'})
}
# レスポンスを作成
item = resp['Item']['item']
return {
'statusCode': 200,
'body': json.dumps({'id': id, 'item': item})
}
テストコードです。
以下の2点を修正しました。
・lambda_handlerのインポートと呼び出し方(app.lambda_handler)
・テスト用のIDの修正(生成コードは{'id': 'test_id'}想定でしたが{'id': '10'}でテストデータを作成したため)
test_handler.py
import json
import unittest
#from show_item import lambda_handler
from show_item import app
class TestShowItem(unittest.TestCase):
def test_success(self):
#event = {'queryStringParameters': {'id': 'test_id'}}
event = {'queryStringParameters': {'id': '10'}}
resp = app.lambda_handler(event, None)
self.assertEqual(resp['statusCode'], 200)
body = resp['body']
data = json.loads(body)
self.assertEqual(data['id'], '10')
self.assertEqual(data['item'], 'test')
def test_missing_id(self):
event = {}
resp = app.lambda_handler(event, None)
self.assertEqual(resp['statusCode'], 400)
def test_item_not_found(self):
event = {'queryStringParameters': {'id': 'invalid_id'}}
resp = app.lambda_handler(event, None)
self.assertEqual(resp['statusCode'], 404)
if __name__ == '__main__':
unittest.main()
なぜかpytestだとうまくいかなかったですが、python -m pytestだと通りました。
$ python -m pytest
======================================================================================= test session starts =======================================================================================
platform linux -- Python 3.8.16, pytest-7.4.3, pluggy-1.3.0
rootdir: /home/ec2-user/environment/bedrock/test_code
plugins: anyio-4.1.0
collected 3 items
tests/unit/test_handler.py ... [100%]
======================================================================================== 3 passed in 0.39s ========================================================================================
template.yamlは完璧でした。
template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Sample SAM Template
Resources:
ShowItemFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: show_item/
Handler: app.lambda_handler
Runtime: python3.9
MemorySize: 128
Timeout: 30
Policies:
- DynamoDBCrudPolicy:
TableName: Test
Events:
GetResource:
Type: Api
Properties:
Path: /show-item
Method: post
TestTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: Test
AttributeDefinitions:
- AttributeName: id
AttributeType: S
KeySchema:
- AttributeName: id
KeyType: HASH
ProvisionedThroughput:
ReadCapacityUnits: 1
WriteCapacityUnits: 1
以下の通りエラーなし。
$ sam validate --region ap-northeast-1 --lint
/home/ec2-user/environment/bedrock/test_code/template.yaml is a valid SAM Template
ビルドもエラーなし。
$ sam build --use-container
Starting Build inside a container
Building codeuri: /home/ec2-user/environment/bedrock/test_code/show_item runtime: python3.9 metadata: {} architecture: x86_64 functions: ShowItemFunction
Fetching public.ecr.aws/sam/build-python3.9:latest-x86_64 Docker container image......
Mounting /home/ec2-user/environment/bedrock/test_code/show_item as /tmp/samcli/source:ro,delegated, inside runtime container
Build Succeeded
Built Artifacts : .aws-sam/build
Built Template : .aws-sam/build/template.yaml
Commands you can use next
=========================
[*] Validate SAM template: sam validate
[*] Invoke Function: sam local invoke
[*] Test Function in the Cloud: sam sync --stack-name {{stack-name}} --watch
[*] Deploy: sam deploy --guided
Running PythonPipBuilder:ResolveDependencies
Running PythonPipBuilder:CopySource
samconfig.tomlは追加で作成しました。
samconfig.toml
version = 0.1
[default]
[default.global.parameters]
stack_name = "test-code"
[default.build.parameters]
cached = true
parallel = true
[default.validate.parameters]
lint = true
[default.deploy.parameters]
capabilities = "CAPABILITY_IAM"
confirm_changeset = true
resolve_s3 = true
[default.package.parameters]
resolve_s3 = true
[default.sync.parameters]
watch = true
[default.local_start_api.parameters]
warm_containers = "EAGER"
[default.local_start_lambda.parameters]
warm_containers = "EAGER"
デプロイもエラーなし。
$ sam deploy --region ap-northeast-1
CloudFormation events from stack operations (refresh every 5.0 seconds)
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ResourceStatus ResourceType LogicalResourceId ResourceStatusReason
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
CREATE_IN_PROGRESS AWS::IAM::Role ShowItemFunctionRole -
CREATE_IN_PROGRESS AWS::DynamoDB::Table TestTable -
CREATE_IN_PROGRESS AWS::DynamoDB::Table TestTable Resource creation Initiated
CREATE_IN_PROGRESS AWS::IAM::Role ShowItemFunctionRole Resource creation Initiated
CREATE_COMPLETE AWS::DynamoDB::Table TestTable -
CREATE_COMPLETE AWS::IAM::Role ShowItemFunctionRole -
CREATE_IN_PROGRESS AWS::Lambda::Function ShowItemFunction -
CREATE_IN_PROGRESS AWS::Lambda::Function ShowItemFunction Resource creation Initiated
CREATE_COMPLETE AWS::Lambda::Function ShowItemFunction -
CREATE_IN_PROGRESS AWS::ApiGateway::RestApi ServerlessRestApi -
CREATE_IN_PROGRESS AWS::ApiGateway::RestApi ServerlessRestApi Resource creation Initiated
CREATE_COMPLETE AWS::ApiGateway::RestApi ServerlessRestApi -
CREATE_IN_PROGRESS AWS::ApiGateway::Deployment ServerlessRestApiDeploymentd002a2810b -
CREATE_IN_PROGRESS AWS::Lambda::Permission ShowItemFunctionGetResourcePermissionProd -
CREATE_IN_PROGRESS AWS::Lambda::Permission ShowItemFunctionGetResourcePermissionProd Resource creation Initiated
CREATE_IN_PROGRESS AWS::ApiGateway::Deployment ServerlessRestApiDeploymentd002a2810b Resource creation Initiated
CREATE_COMPLETE AWS::Lambda::Permission ShowItemFunctionGetResourcePermissionProd -
CREATE_COMPLETE AWS::ApiGateway::Deployment ServerlessRestApiDeploymentd002a2810b -
CREATE_IN_PROGRESS AWS::ApiGateway::Stage ServerlessRestApiProdStage -
CREATE_IN_PROGRESS AWS::ApiGateway::Stage ServerlessRestApiProdStage Resource creation Initiated
CREATE_COMPLETE AWS::ApiGateway::Stage ServerlessRestApiProdStage -
CREATE_COMPLETE AWS::CloudFormation::Stack test-code -
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Successfully created/updated stack - test-code in ap-northeast-1
クエリパラメータの想定でしたが、クエリストリングで実行できました。
$ curl -X POST https://******.execute-api.ap-northeast-1.amazonaws.com/Prod/show-item?id=10
{"id": "10", "item": "test"}
細かい修正がいくつかありましたが、いずれもこちらの指定不足でした。
フィードバックをすればすぐに修正してくれそうです。
そもそも、適当に作った設計書で、よくここまで書いてくれたと思います。
今回試した内容は、以下のGitHubにコミットしています。