[boto3] Sessionのリージョンを切り替える。ついでにSwitchRoleも。


やりたいこと

boto3 では

session = boto3.Session(region_name="<region-name>")

上の例のようなコードでリージョンを指定してセッションを作成することが出来ますが、一度作成したSessionのリージョンを変更することはできないようです。

実際に使うときは下の例のように

client   = session.client("<service-name>", region_name="<region-name>")
resource = session.resource("<service-name>", region_name="<region-name>")

サービス別の client や resource を作成するときに region_name を指定できるので、そこで指定すればリージョン別に処理をすることは出来ます。
出来ますが、毎回指定するのが面倒で且つ指定漏れ(コーディングミス)の懸念があるため、根っこに当たるSessionでリージョンを変更したい場面もあります。私はありました。
リージョンを頻繁に切り替えたりする場合や、それに 加えてSwitchRole (AssumeRole)も絡んでくると、更に面倒なことになってきます。

解決方法

指定セッションの情報を引き継いで新たにセッションを生成するファンクションを作りました。
リージョン変更が主目的でしたがSwitchRoleのコードも毎回書くと面倒なので、1つにまとめました。

コード

snippet
import boto3

def get_aws_session(base_session=None, region_name=None, **assume_role_args):

    # デフォルト値の調整
    if base_session is None:
        base_session = boto3.Session()
    if region_name is None:
        region_name = base_session.region_name

    # SwitchRole Session生成
    if len(assume_role_args) > 0:
        response  = base_session.client("sts").assume_role(**assume_role_args)
        session_args = {
            "aws_access_key_id"     : response['Credentials']['AccessKeyId'],
            "aws_secret_access_key" : response['Credentials']['SecretAccessKey'],
            "aws_session_token"     : response['Credentials']['SessionToken'],
            "region_name"           : region_name,
        }
        return boto3.Session(**session_args)

    # Session 生成(リージョン切り替えのみ)
    credentials = base_session.get_credentials()
    if credentials.method in ["explicit", "env"]:
        # AccessKey, SecretKey[, SessionToken] 指定の場合
        session_args = {
            "aws_access_key_id"     : credentials.access_key,
            "aws_secret_access_key" : credentials.secret_key,
            "region_name"           : region_name,
        }
        if credentials.token is not None:
            session_args["aws_session_token"] = credentials.token
        return boto3.Session(**session_args)

    else:
        # Profile 指定の場合
        session_args = {
            "profile_name"     : base_session.profile_name,
            "region_name"      : region_name,
        }
        return boto3.Session(**session_args)

パラメータ

パラメータ名 説明
base_session 基底となるセッション。
このセッションをもとに、リージョンを変更したりSwitchRoleしたり。
未指定の場合はboto3.Session()
region_name 変更するリージョン。
未指定の場合は基底セッションのものを引き継ぐ。
**assume_role_args AssumeRoleコマンドに引き渡す任意のキーワード引数
使用できるパラメータは [boto3 Docs - STS.assume_role] の項を参照。

使用例

sample
# 現在のセッションをベースにリージョンだけ切り替える。
new_session_1 = get_aws_session(region_name="us-east-1")


# 現在のセッションをベースにSwitchRoleする。
new_session_2 = get_aws_session(
    RoleArn='arn:aws:iam::<account-id>:role/<role-name>',
    RoleSessionName='switch-role-test'
)


# SwitchRoleしたセッションから更にSwitchRole(多段SwitchRole)
# ついでにリージョンも変更。
new_session_3 = get_aws_session(
    base_session=new_session_2,
    region_name="eu-central-1",
    RoleArn='arn:aws:iam::<account-id>:role/<role-name>',
    RoleSessionName='<session-name>'
)


# おまけ:現在のセッションから何も変えない。boto3.Session()と同義(無意味)
new_session_4 = get_aws_session()


ちょっと改良

  • boto3.Session.session って、アカウントIDを持ってないんですよね。不便ですね。
    なので戻り値の sessionaccount_id を追加して参照可能に。
    ただし結構遅くなる。
  • SwitchRoleもRegion切り替えも実施しない場合は現在のセッションを返すように。
def get_aws_session(base_session=None, region_name=None, **assume_role_args):

    # デフォルト値の調整
    if base_session is None:
        base_session = boto3.Session()
    if region_name is None:
        region_name = base_session.region_name

    # session生成処理
    credentials = base_session.get_credentials()

    if len(assume_role_args) > 0:
        # SwitchRole Session生成
        response  = base_session.client("sts").assume_role(**assume_role_args)
        session_args = {
            "aws_access_key_id"     : response['Credentials']['AccessKeyId'],
            "aws_secret_access_key" : response['Credentials']['SecretAccessKey'],
            "aws_session_token"     : response['Credentials']['SessionToken'],
            "region_name"           : region_name,
        }
        session = boto3.Session(**session_args)

    else:
        # リージョン切り替えのみ

        if ( base_session.region_name == region_name ):
            # 現在と同一であれば処理しない。
            session = base_session

        else:
            # リージョン切り替え実施

            if credentials.method in ["explicit", "env"]:
                # AccessKey, SecretKey[, SessionToken] 指定の場合
                session_args = {
                    "aws_access_key_id"     : credentials.access_key,
                    "aws_secret_access_key" : credentials.secret_key,
                    "region_name"           : region_name,
                }
                if credentials.token is not None:
                    session_args["aws_session_token"] = credentials.token
                session = boto3.Session(**session_args)

            else:
                # Profile 指定の場合
                session_args = {
                    "profile_name"     : base_session.profile_name,
                    "region_name"      : region_name,
                }
                session = boto3.Session(**session_args)

    session.account_id = session.client("sts").get_caller_identity()["Account"]
    return session

↓ 使い方

new_session_1 = get_aws_session(region_name="us-east-1")
print(new_session_1.account_id)