def _load_materials()

in src/dynamodb_encryption_sdk/material_providers/store/meta.py [0:0]
30 lines of code
5 McCabe index (conditional complexity)

    def _load_materials(self, material_name, version):
        # type: (Text, int) -> Tuple[JceNameLocalDelegatedKey, JceNameLocalDelegatedKey]
        """Load materials from table.

        :returns: Materials loaded into delegated keys
        :rtype: tuple(JceNameLocalDelegatedKey)
        """
        _LOGGER.debug('Loading material "%s" version %d from MetaStore table', material_name, version)
        key = {MetaStoreAttributeNames.PARTITION.value: material_name, MetaStoreAttributeNames.SORT.value: version}
        response = self._encrypted_table.get_item(Key=key)
        try:
            item = response["Item"]
        except KeyError:
            raise InvalidVersionError('Version not found: "{}#{}"'.format(material_name, version))

        try:
            encryption_key_kwargs = dict(
                key=item[MetaStoreAttributeNames.ENCRYPTION_KEY.value].value,
                algorithm=item[MetaStoreAttributeNames.ENCRYPTION_ALGORITHM.value],
                key_type=EncryptionKeyType.SYMMETRIC,
                key_encoding=KeyEncodingType.RAW,
            )
            signing_key_kwargs = dict(
                key=item[MetaStoreAttributeNames.INTEGRITY_KEY.value].value,
                algorithm=item[MetaStoreAttributeNames.INTEGRITY_ALGORITHM.value],
                key_type=EncryptionKeyType.SYMMETRIC,
                key_encoding=KeyEncodingType.RAW,
            )
        except KeyError:
            raise Exception("Invalid record")

        # need to handle if the material type version is not in the item
        # https://github.com/aws/aws-dynamodb-encryption-python/issues/140
        if item[MetaStoreAttributeNames.MATERIAL_TYPE_VERSION.value] != MetaStoreValues.MATERIAL_TYPE_VERSION.value:
            raise InvalidVersionError(
                'Unsupported material type: "{}"'.format(item[MetaStoreAttributeNames.MATERIAL_TYPE_VERSION.value])
            )

        encryption_key = JceNameLocalDelegatedKey(**encryption_key_kwargs)
        signing_key = JceNameLocalDelegatedKey(**signing_key_kwargs)
        return encryption_key, signing_key