"""Dynamo DB module.""" import os import boto3 from boto3 import resource as DynamodbResource from boto3.dynamodb import table as DynamodbTable from botocore import config as botocore_config from moneyhub.constants.constants import Environment DYNAMODB_TABLE_SUCCESS = '{env}_statement_attachments_success' DDB_WRITE_MAX_ATTEMPTS = int(os.environ.get('DDB_WRITE_MAX_ATTEMPTS', 10)) def get_dynamodb(max_attempts=None) -> DynamodbResource: """Get Boto DynamoDB resource. Args: max_attempts (int): optional max retry attempts Returns: boto3.resources.factory.dynamodb.ServiceResource: DynamoDB resource """ if max_attempts is None: max_attempts = DDB_WRITE_MAX_ATTEMPTS config_instance = botocore_config.Config( retries={'max_attempts': max_attempts}) return boto3.resource( 'dynamodb', config=config_instance) def get_dynamodb_table(env: Environment) -> DynamodbTable: """Get the DynamoDB table. Args: env (str): the environment of the table. Returns: boto3.resources.factory.dynamodb.Table: DynamoDB table """ db = get_dynamodb() return db.Table(DYNAMODB_TABLE_SUCCESS.format(env=env.value))