def get_boto_client(client_type):
aws_region = os.environ[AWS_REGION]
global boto_client_cache
if boto_client_cache:
# check if clients are stale and clear global cache if they are
client_age = time.time() - boto_client_cache[TIMESTAMP]
client_expiration_age = int(os.getenv(CLIENT_EXPIRATION_AGE)) # override env var to change refresh interval
if client_age > client_expiration_age:
boto_client_cache.clear()
if not boto_client_cache:
# create clients if the global cache is empty
boto_client_cache[TIMESTAMP] = time.time()
boto_client_cache[DDB] = __get_dynamodb_client(aws_region)
try:
client = boto_client_cache[client_type]
except KeyError:
logger.exception("Client is missing from the global boto cache")
client = None
return client
def __get_dynamodb_client(aws_region):
boto_config = __get_boto_config("ddb")
ddb_client = boto3.client("dynamodb", region_name=aws_region, config=boto_config)
return ddb_client
def __get_boto_config(client_type):
max_retries = 5 # 8-10 retries needed to trigger DDB auto scaling
if client_type == "ddb":
max_retries = 10
boto_config = botocore.config.Config(
connect_timeout=CONNECT_TIMEOUT, read_timeout=READ_TIMEOUT, retries={"max_attempts": max_retries}
)
return boto_config