在 AWS 中存储从第三方 API 生成的访问令牌的最佳方式是什么

Question

我目前正在运行一个使用 GoogleAdsManager API 生成报告的本地脚本。在运行脚本之前，我创建了 json 格式的新服务帐户密钥作为密钥类型以及 ~/googleads.yaml.

Here's 开发指南。

但是，我想将此脚本安排在 (AWS Glue)。

这是示例脚本，我目前面临的问题是：

如何从 aws 调用此方法 ad_manager.AdManagerClient.LoadFromStorage()？我已将凭据（JSON 和 YAML）存储在 AWS Secrets Manager

中

from googleads import ad_manager, oauth2
import tempfile
import _locale

_locale._getdefaultlocale = (lambda *args: ['en_US', 'UTF-8'])

ad_unit_id = XXXXXXXXXX

def generate_ad_impressions(client):
    # Initialize appropriate service.
    report_service = client.GetService("ReportService", 
    version="v202108")
    # Initialize a DataDownloader.
    report_downloader = client.GetDataDownloader(version="v202108")
    # Create statement object to filter for an order.
    statement = (
        ad_manager.StatementBuilder(version="v202108")
        .Where("PARENT_AD_UNIT_ID = :id")
        .WithBindVariable("id", mbns_aa_vod_ad_unit_id)
        .Limit(None)  # No limit/offset for reports
        .Offset(None)
    )
    report_job = {
        "reportQuery": {
        "dimensions": ["DATE", "HOUR"],
        "columns": [
            "AD_SERVER_IMPRESSIONS",
        ],
        "dateRangeType": "TODAY",
        "startDate": {
            "year": "2022",
            "month": "1",
            "day": "25"
        },
        "endDate": {
            "year": "2022",
            "month": "1",
            "day": "25"
        },
        "statement": statement.ToStatement(),
      }
    }
    try:
        # Run the report and wait for it to finish.
        report_job_id = report_downloader.WaitForReport(report_job)
    except:
        print("Failed to generate report.")
        # Change to your preferred export format.
        export_format = "CSV_DUMP"
        # report_file = 
        tempfile.NamedTemporaryFile(suffix=".csv.gz", 
        delete=False)
    with open('ad_unit_report.csv.gz', mode='wb') as report_file:
        # Download report data.
        report_downloader.DownloadReportToFile(report_job_id, 
        export_format, report_file)
        report_file.close()

    # Download report data.
    downloaded_report = 
    report_downloader.DownloadReportToFile(report_job_id, 
    export_format, report_file)

    report_file.close()
    print('success!')

if __name__ == '__main__':
   ad_manager_client = 
   ad_manager.AdManagerClient.LoadFromStorage('path_to_yaml_file')
   generate_ad_impressions(ad_manager_client)

Answer 1

解决方案应该是这样的：

import boto3
from botocore.exceptions import ClientError
from googleads import ad_manager, oauth2
import tempfile
import _locale

_locale._getdefaultlocale = (lambda *args: ['en_US', 'UTF-8'])

ad_unit_id = XXXXXXXXXX

def get_secret():
    secret_name = "MySecret"
    region_name = "us-west-2"

    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name,
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            print("The requested secret " + secret_name + " was not found")
        elif e.response['Error']['Code'] == 'InvalidRequestException':
            print("The request was invalid due to:", e)
        elif e.response['Error']['Code'] == 'InvalidParameterException':
            print("The request had invalid params:", e)
        elif e.response['Error']['Code'] == 'DecryptionFailure':
            print("The requested secret can't be decrypted using the provided KMS key:", e)
        elif e.response['Error']['Code'] == 'InternalServiceError':
            print("An error occurred on service side:", e)
    else:
        # Secrets Manager decrypts the secret value using the associated KMS CMK
        # Depending on whether the secret was a string or binary, only one of these fields will be populated
        if 'SecretString' in get_secret_value_response:
            text_secret_data = get_secret_value_response['SecretString']
            return text_secret_data
        else:
            binary_secret_data = get_secret_value_response['SecretBinary']
            return binary_secret_data

def generate_ad_impressions(client):
    # Initialize appropriate service.
    report_service = client.GetService("ReportService", 
    version="v202108")
    # Initialize a DataDownloader.
    report_downloader = client.GetDataDownloader(version="v202108")
    # Create statement object to filter for an order.
    statement = (
        ad_manager.StatementBuilder(version="v202108")
        .Where("PARENT_AD_UNIT_ID = :id")
        .WithBindVariable("id", mbns_aa_vod_ad_unit_id)
        .Limit(None)  # No limit/offset for reports
        .Offset(None)
    )
    report_job = {
        "reportQuery": {
        "dimensions": ["DATE", "HOUR"],
        "columns": [
            "AD_SERVER_IMPRESSIONS",
        ],
        "dateRangeType": "TODAY",
        "startDate": {
            "year": "2022",
            "month": "1",
            "day": "25"
        },
        "endDate": {
            "year": "2022",
            "month": "1",
            "day": "25"
        },
        "statement": statement.ToStatement(),
      }
    }
    try:
        # Run the report and wait for it to finish.
        report_job_id = report_downloader.WaitForReport(report_job)
    except:
        print("Failed to generate report.")
        # Change to your preferred export format.
        export_format = "CSV_DUMP"
        # report_file = 
        tempfile.NamedTemporaryFile(suffix=".csv.gz", 
        delete=False)
    with open('ad_unit_report.csv.gz', mode='wb') as report_file:
        # Download report data.
        report_downloader.DownloadReportToFile(report_job_id, 
        export_format, report_file)
        report_file.close()

    # Download report data.
    downloaded_report = 
    report_downloader.DownloadReportToFile(report_job_id, 
    export_format, report_file)

    report_file.close()
    print('success!')

if __name__ == '__main__':
   ad_manager_client = 
   ad_manager.AdManagerClient.LoadFromString(get_secret())
   generate_ad_impressions(ad_manager_client)

此处函数 get_secret 从 AWS Secrets Manager 中获取 YAML 字符串，存储为 MySecret 在区域 us-west-2 中。随后使用 YAML 字符串通过函数 LoadFromString.

创建 Google Ad Manager 客户端

请注意，我没有对此进行测试，因为我没有 Google API 密钥和 ID。

在 AWS 中存储从第三方 API 生成的访问令牌的最佳方式是什么

What's the best way to store access token generated from third party API in AWS

python

amazon-web-services

aws-glue

aws-secrets-manager

google-ads-api