Skip to content

Commit

Permalink
Simplify host account initalization (#80)
Browse files Browse the repository at this point in the history
Initialize the assuming session before initializing the clients.

Inline each client initialization. A little repeated structure here is easier
to read than the complex type signatures of the deleted helper functions.

In a future change the STS client will take extra parameters to support version
2 STS tokens and opt-in regions.

This paves the way to a solution for #77.
  • Loading branch information
iainelder authored Nov 26, 2023
1 parent 091ce34 commit 1c61504
Showing 1 changed file with 17 additions and 47 deletions.
64 changes: 17 additions & 47 deletions botocove/cove_host_account.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,12 @@
import logging
import re
from functools import lru_cache
from typing import (
Any,
Dict,
Iterable,
List,
Literal,
Optional,
Sequence,
Set,
Tuple,
Union,
)

import boto3
from typing import Dict, Iterable, List, Optional, Sequence, Set, Tuple

from boto3.session import Session
from botocore.config import Config
from botocore.exceptions import ClientError
from mypy_boto3_organizations.client import OrganizationsClient
from mypy_boto3_organizations.type_defs import AccountTypeDef
from mypy_boto3_sts.client import STSClient
from mypy_boto3_sts.type_defs import PolicyDescriptorTypeTypeDef

from botocove.cove_types import CoveSessionInformation
Expand Down Expand Up @@ -52,8 +38,21 @@ def __init__(

self.thread_workers = thread_workers

self.sts_client = self._get_boto3_sts_client(assuming_session)
self.org_client = self._get_boto3_org_client(assuming_session)
if assuming_session:
logger.info(f"Using provided Boto3 session {assuming_session}")
else:
logger.info("No Boto3 session argument: using credential chain")
assuming_session = Session()

self.sts_client = assuming_session.client(
service_name="sts",
config=Config(max_pool_connections=self.thread_workers),
)

self.org_client = assuming_session.client(
service_name="organizations",
config=Config(max_pool_connections=self.thread_workers),
)

caller_id = self.sts_client.get_caller_identity()
self.host_account_id = caller_id["Account"]
Expand Down Expand Up @@ -147,35 +146,6 @@ def _generate_account_sessions(self) -> Iterable[CoveSessionInformation]:
Result=None,
)

def _get_boto3_client(
self,
clientname: Union[Literal["organizations"], Literal["sts"]],
assuming_session: Optional[Session],
) -> Any:
if assuming_session:
logger.info(f"Using provided Boto3 session {assuming_session}")
return assuming_session.client(
service_name=clientname,
config=Config(max_pool_connections=self.thread_workers),
)
logger.info("No Boto3 session argument: using credential chain")
return boto3.client(
service_name=clientname,
config=Config(max_pool_connections=self.thread_workers),
)

def _get_boto3_org_client(
self, assuming_session: Optional[Session]
) -> OrganizationsClient:
client: OrganizationsClient = self._get_boto3_client(
"organizations", assuming_session
)
return client

def _get_boto3_sts_client(self, assuming_session: Optional[Session]) -> STSClient:
client: STSClient = self._get_boto3_client("sts", assuming_session)
return client

def _resolve_target_accounts(self, target_ids: Optional[List[str]]) -> Set[str]:
accounts_to_ignore = self._gather_ignored_accounts()
logger.info(f"Ignoring account IDs: {accounts_to_ignore=}")
Expand Down

0 comments on commit 1c61504

Please sign in to comment.