def get_replicated_write_pricing()

in ddbtools/pricing.py [0:0]

• 31 lines of code
• 6 McCabe index (conditional complexity)

    def get_replicated_write_pricing(self, region_code: str) -> dict:
        """Get DynamoDB replicated write (for global tables) pricing for a given region."""
        replicated_writes_pricing = {} 

        response = self.pricing_client.get_products(
            ServiceCode='AmazonDynamoDB',
            Filters=[{'Type': 'TERM_MATCH',
                      'Field': 'productFamily',
                      'Value': 'DDB-Operation-ReplicatedWrite'},
                     {'Type': 'TERM_MATCH',
                      'Field': 'regionCode',
                      'Value': region_code}
            ],
            FormatVersion='aws_v1',
            MaxResults=100
        )
        price_list = response['PriceList']

        for entry in price_list:
            product = json.loads(entry)
            product_group = product['product']['attributes']['group']
            offer = product['terms']['OnDemand'].popitem()
            offer_terms = offer[1]
            price_dimensions = offer_terms['priceDimensions']

            for price_dimension_code in price_dimensions:
                price_terms = price_dimensions[price_dimension_code]
                price_per_unit = price_terms['pricePerUnit']['USD']
                price = Decimal(price_per_unit)

                # Regions with free tier pricing will have an initial entry set to zero; skip this
                if price != 0:
                    if product_group == 'DDB-ReplicatedWriteUnits':
                        replicated_writes_pricing[constants.REPLICATED_STD_WCU_PRICING] = price
                    elif product_group == 'DDB-ReplicatedWriteUnitsIA':
                        replicated_writes_pricing[constants.REPLICATED_IA_WCU_PRICING] = price

        return replicated_writes_pricing