from typing import Dict, Union
from fastapi import APIRouter, Header, status
from config import get_settings
from schemas.sample_request import SampleRequest
import json
import boto3
from botocore.exceptions import ClientError
router = APIRouter()
settings = get_settings()
# HTTP POSTメソッドを使ってリクエストを受け取り、それに応じてレスポンスを返すエンドポイントを定義
# エンドポイントのURLは、/sample_apiとなっている
@router.post('/sample_api')
def sample_api(
request: SampleRequest,
x_api_key: Union[str, None] = Header(default=None)) -> Dict:
# AWS APIキーを取得するためのJSONファイルを読み込む
# settings.app_pathは、このFastAPIアプリケーションのルートディレクトリを表す
with open(settings.app_path / 'aws_api_key.json', 'r') as f:
key_json: Dict = json.load(f)
# secret_nameとregion_nameは、AWS Secrets Managerのシークレットを取得するために必要な情報
secret_name = key_json.get('secret_name')
region_name = key_json.get('region_name')
# company_api_keyは、会社コードに応じたシークレットを特定するためのキーとして使用
company_api_key = f"{request.company_cd}_api_key"
print(company_api_key) # この結果はリクエストボディに含まれるcompany_cdフィールドの値Foo01_api_keyが出力される
# 会社コードに応じたAWS Secrets Managerからのシークレットを取得する
try:
# get_secret()関数を呼び出して、AWS Secrets Managerからシークレットを取得
# return(jsonDict[company_api_key])とされているため、取得したシークレットのうち、company_api_keyで指定された値を返される。
# 返された値をsecret: strに代入。secret変数には、company_api_keyで指定された値(hogehoge)が格納される
secret: str = get_secret(secret_name, region_name, company_api_key)
print(secret) # この結果はhogehoge（シークレットキーの値）が出力される
# 会社コードに応じたシークレットキーが存在しなかった場合は、エラーを返す
except KeyError:
return('会社コードによるキーエラーの場合に返す内容を実装する')
# X_API_KEYが一致しない場合は、エラーを返す
if x_api_key != secret:
return('X_API_KEYエラーの場合に返す内容を実装する')
# 他にエラーチェックを記載する場合は続けて記載or実装する
# if request.request_id == None: など...
return ('レスポンスボディの内容をここに記載')
# AWSシークレットマネージャからキーを取得する
def get_secret(secret_name, region_name, company_api_key) -> str:
# AWS SDK for Python (Boto3)を使用して、シークレットマネージャーのクライアントを作成
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
# 指定されたsecret_nameでシークレットを取得
# withステートメントによりaws_api_key.jsonで指定したシークレット(sample_secret)が取得されている
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
print(secret_name) # この結果はsample_secretが出力される
except ClientError as e:
raise e
# 例えば、jsonファイルのsecret_nameの値を適当にすると、以下のエラーが出る
# botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the GetSecretValue operation: Secrets Manager can't find the specified secret.
# シークレットを文字列として読み込み、JSONフォーマットに変換
secret = get_secret_value_response['SecretString']
jsonDict: Dict = json.loads(secret)
print(get_secret_value_response) # この結果はARNから先ほど登録したAPIキーの情報までを持っている辞書型の変数が出力される
print(secret) # この結果は['SecretString']に該当する{"Foo01_api_key":"hogehoge"}が出力される
print(company_api_key) # この結果はFoo01_api_keyが出力される
# シークレットのJSONデータから、指定されたシークレットキーに対応する値を取得する
# 今回はcompany_api_keyで指定されたキーに対応する値を取得し、その値を返す
return(jsonDict[company_api_key]) # 辞書型の変数からcompany_api_keyで指定された値(hogehoge)を取り出して返す
