import os import json import botocore import boto3 import api_common from aws_lambda_powertools import Logger, Tracer from aws_lambda_powertools.utilities.typing import LambdaContext # QuickSight ã®ã‚µã‚¤ãƒ³ã‚¢ãƒƒãƒ—ã‚’è¡Œã£ãŸãƒªãƒ¼ã‚¸ãƒ§ãƒ³ã‚’環境変数ã‹ã‚‰å–å¾— QUICKSIGHT_IDENTITY_REGION = os.environ['QUICKSIGHT_IDENTITY_REGION'] QUICKSIGHT_FEDERATION_ROLE_NAME = os.environ['QUICKSIGHT_FEDERATION_ROLE_NAME'] # QuickSight ユーザーã®åå‰ç©ºé–“を環境変数ã‹ã‚‰å–å¾— QUICKSIGHT_USER_NAMESPACE = os.environ['QUICKSIGHT_USER_NAMESPACE'] # DynamoDB テーブル㮠ARN を環境変数ã‹ã‚‰å–å¾— DYNAMODB_TABLE_NAME_DASHBOARDS = os.environ['DYNAMODB_TABLE_NAME_DASHBOARDS'] # ãƒã‚°ã¨ãƒˆãƒ¬ãƒ¼ã‚¹ã®æ©Ÿèƒ½ã‚’åˆæœŸåŒ– logger = Logger() tracer = Tracer() # STS ã®ã‚¯ãƒ©ã‚¤ã‚¢ãƒ³ãƒˆã‚’åˆæœŸåŒ– quicksight = boto3.client('quicksight') dynamodb = boto3.client('dynamodb') # ワークフãƒãƒ¼ã®å‡ºåŠ›ã‚’å¯è¦–化ã™ã‚‹ãƒ€ãƒƒã‚·ãƒ¥ãƒœãƒ¼ãƒ‰ã‚’扱ㆠAPI を実装ã—㟠Lambda 関数ã®ãƒãƒ³ãƒ‰ãƒ© # CloudWatch Logs 㨠X-Ray ã«ã‚ˆã‚‹ãƒã‚°ã¨ãƒˆãƒ¬ãƒ¼ã‚¹ã‚’有効化 @tracer.capture_lambda_handler @logger.inject_lambda_context(log_event=True) def handler(event: dict, context: LambdaContext) -> dict: # REST API ã«æŒ‡å®šã•ã‚ŒãŸãƒ‘スã¨ã‚¯ã‚¨ãƒªæ–‡å—列をå–å¾— pathParams = event.get('pathParameters') if not pathParams: # ãƒ‘ã‚¹æƒ…å ±ãŒãªã‘れ㰠400 Bad Request ã¨ã™ã‚‹ return { 'statusCode': 400, 'headers': api_common.CORS_HEADERS, } runId = pathParams.get('runId') visualizationId = pathParams.get('visualizationId') if not runId or not visualizationId: # パスã«å®Ÿè¡Œ ID ã¨å¯è¦–化 ID ãŒå«ã¾ã‚Œã¦ã„ãªã‘れ㰠404 Not Found ã¨ã™ã‚‹ return { 'statusCode': 404, 'headers': api_common.CORS_HEADERS, } headers = event['headers'] try: accountId = event['requestContext']['accountId'] userId = event['requestContext']['authorizer']['claims']['sub'] userArn = f'arn:aws:quicksight:{QUICKSIGHT_IDENTITY_REGION}:{accountId}:user/{QUICKSIGHT_USER_NAMESPACE}/{QUICKSIGHT_FEDERATION_ROLE_NAME}/{userId}' # パスã«å‡ºåŠ›ãƒ•ã‚¡ã‚¤ãƒ«ã®ãƒ‘スãŒå«ã¾ã‚Œã¦ã„ãªã‹ã£ãŸã‚‰ã€å…¨ã¦ã®å‡ºåŠ›ãƒ•ã‚¡ã‚¤ãƒ«ã®ä¸€è¦§ã‚’返㙠return handle_get_dashboard(accountId, userArn, runId, visualizationId, headers) except botocore.exceptions.ClientError as err: statusCode = err.response['ResponseMetadata']['HTTPStatusCode'] code = err.response['Error']['Code'] message = err.response['Error']['Message'] logger.exception(f'{code}: {message}') # AWS ã® API ã‹ã‚‰ã‚¨ãƒ©ãƒ¼ãƒ¬ã‚¹ãƒãƒ³ã‚¹ãŒè¿”ã•ã‚ŒãŸã‚‰ã€ãã®å†…容ã«æº–ã˜ãŸã‚¨ãƒ©ãƒ¼ã‚³ãƒ¼ãƒ‰ã¨ãƒ¡ãƒƒã‚»ãƒ¼ã‚¸ã‚’返㙠return { 'statusCode': statusCode, 'headers': { 'Content-Type': 'application/json', **api_common.CORS_HEADERS, }, 'body': json.dumps({ 'code': code, 'message': message, }, default=api_common.default_serializer), } except (botocore.exceptions.BotoCoreError, ValueError) as err: code = type(err).__name__ message = str(err) logger.exception(f'{code}: {message}') # ãã®ä»–ã®ã‚¨ãƒ©ãƒ¼ãŒç™ºç”Ÿã—ãŸã‚‰ã€ã‚¨ãƒ©ãƒ¼ãƒ¡ãƒƒã‚»ãƒ¼ã‚¸ã¨å…±ã« 400 Bad Request を返㙠return { 'statusCode': 400, 'headers': { 'Content-Type': 'application/json', **api_common.CORS_HEADERS, }, 'body': json.dumps({ 'code': code, 'message': message, }, default=api_common.default_serializer), } # ダッシュボードã®åŸ‹ã‚込㿠URL ã«é–¢ã™ã‚‹å‡¦ç†ã‚’行ㆠdef handle_get_dashboard(accountId: str, userArn: str, runId: str, visualizationId: str, headers: dict) -> dict: # DynamoDB ã® Dashboards テーブルã‹ã‚‰ãƒ€ãƒƒã‚·ãƒ¥ãƒœãƒ¼ãƒ‰ã® ID ã‚’å–å¾—ã™ã‚‹ response = dynamodb.get_item( TableName=DYNAMODB_TABLE_NAME_DASHBOARDS, Key=api_common.dict_to_dynamodb({ 'runId': runId, 'visualizationId': visualizationId, }), ) item = api_common.dict_from_dynamodb(response.get('Item')) dashboardId = item.get('dashboardId') if item else None if not dashboardId: # ダッシュボード ID ãŒè¨˜éŒ²ã•ã‚Œã¦ã„ãªã‘れ㰠404 Not Found ã¨ã™ã‚‹ return { 'statusCode': 404, 'headers': api_common.CORS_HEADERS, } # ユーザーã«å¯¾å¿œã™ã‚‹ QuickSight ユーザーã®æ¨©é™ã§ã€ãƒ€ãƒƒã‚·ãƒ¥ãƒœãƒ¼ãƒ‰ã‚’ Web アプリã«åŸ‹ã‚込むãŸã‚ã® URL ã‚’å–å¾—ã—ã¦è¿”ã™ origin = headers['origin'] response = quicksight.generate_embed_url_for_registered_user( AwsAccountId=accountId, UserArn=userArn, ExperienceConfiguration={ 'Dashboard': { 'InitialDashboardId': dashboardId, 'FeatureConfigurations': { 'StatePersistence': { 'Enabled': True, }, }, }, }, AllowedDomains=[ origin, ], ) embedUrl = response['EmbedUrl'] return { 'statusCode': 200, 'headers': { 'Content-Type': 'application/json', **api_common.CORS_HEADERS, }, 'body': embedUrl, }