Source code for idem_aws.exec.aws.neptune.db_cluster_parameter_group

"""Exec module for managing Amazon Neptune DB Cluster Parameter Group."""
from typing import Dict

__func_alias__ = {"list_": "list"}

RESOURCE_TYPE = "aws.neptune.db_cluster_parameter_group"


[docs]async def get( hub, ctx, name: str, resource_id: str = None, ) -> Dict: """Retrieves the specified AWS Neptune DB Cluster Parameter Group. It also fetches the db cluster parameters for the supplied resource_id. Filters is not supported yet as per documentation for this resource on AWS. It returns idem object representation for db_cluster_parameter_group found if resource_id was specified. If resource_id wasn't specified, it returns the first resource found. Args: name(str): An Idem name of the resource. resource_id(str, Optional): DBClusterParameterGroupName of the resource in AWS. .. note:: At the time of writing this module, filters parameter is not supported for this resource as per AWS documentation so idem-aws has omitted it. Returns: Dict[bool, list, dict or None]: result(bool): Whether the result of the function has been successful (``True``) or not (``False``). comment(list): A list of messages. ret(dict or None): The AWS Neptune DB Cluster Parameter Group in "present" format. Examples: Calling this exec module function from the cli: .. code-block:: bash idem exec aws.neptune.db_cluster_parameter_group.get name="idem_name" resource_id="resource_id" Calling this exec module function from within a state module in pure python: .. code-block:: python async def state_function(hub, ctx, name, resource_id, **kwargs): ret = await hub.exec.aws.neptune.db_cluster_parameter_group.get( ctx, name=name, resource_id=resource_id ) """ result = dict(comment=[], ret=None, result=True) ret = await hub.tool.aws.neptune.db_cluster_parameter_group.search_raw( ctx=ctx, name=name, resource_id=resource_id ) if not ret["result"]: if "ParameterGroupNotFound" in str(ret["comment"]): result["comment"].append( hub.tool.aws.comment_utils.get_empty_comment( resource_type=RESOURCE_TYPE, name=name ) ) result["comment"] += list(ret["comment"]) return result result["comment"] += list(ret["comment"]) result["result"] = False return result if not ret["ret"]["DBClusterParameterGroups"]: result["comment"].append( hub.tool.aws.comment_utils.get_empty_comment( resource_type=RESOURCE_TYPE, name=name ) ) return result resource = ret["ret"]["DBClusterParameterGroups"][0] resource_id = resource.get("DBClusterParameterGroupName") if len(ret["ret"]["DBClusterParameterGroups"]) > 1: result["comment"].append( hub.tool.aws.comment_utils.find_more_than_one( resource_type=RESOURCE_TYPE, resource_id=resource_id ) ) arn = resource["DBClusterParameterGroupArn"] # get parameters parameters_raw = await hub.exec.boto3.client.neptune.describe_db_cluster_parameters( ctx=ctx, DBClusterParameterGroupName=resource_id ) if not parameters_raw["result"]: result["result"] = False result["comment"].append(parameters_raw["comment"]) return result parameters_raw = parameters_raw.get("ret") parameters = parameters_raw.get("Parameters") # get tags tags = await hub.tool.aws.neptune.tag.get_tags_for_resource(ctx, resource_arn=arn) if not tags["result"]: result["result"] = False result["comment"].append(tags["comment"]) return result tags = tags["ret"] result[ "ret" ] = hub.tool.aws.neptune.conversion_utils.convert_raw_db_cluster_parameter_group_to_present( raw_resource=resource, idem_resource_name=name, tags=tags, parameters=parameters ) return result
[docs]async def list_(hub, ctx, name: str = None) -> Dict: """Use a list of un-managed neptune db cluster parameter groups as a data-source. Args: name (str, Optional): The name of idem state. .. note:: At the time of writing this module, filters parameter is not supported for this resource as per AWS documentation so idem-aws has omitted it. Returns: Dict[bool, list, dict or None]: result(bool): Whether the result of the function has been successful (``True``) or not (``False``). comment(list): A list of messages. ret(dict or None): The AWS Neptune DB Cluster Parameter Group in "present" format. Examples: Calling this exec module function from the cli: .. code-block:: bash idem exec aws.neptune.db_cluster_parameter_group.list name=name Calling this exec module function from within a state module in pure python: .. code-block:: python async def state_function(hub, ctx, name, **kwargs): ret = await hub.exec.aws.neptune.db_cluster_parameter_group.list( ctx, name=name ) """ result = dict(comment=[], ret=[], result=True) ret = await hub.tool.aws.neptune.db_cluster_parameter_group.search_raw( ctx=ctx, name=name ) if not ret["result"]: result["comment"] += list(ret["comment"]) result["result"] = False return result if not ret["ret"]["DBClusterParameterGroups"]: result["comment"].append( hub.tool.aws.comment_utils.get_empty_comment( resource_type=RESOURCE_TYPE, name=name ) ) return result for db_cluster_parameter_group in ret["ret"]["DBClusterParameterGroups"]: db_cluster_parameter_group_name = db_cluster_parameter_group.get( "DBClusterParameterGroupName" ) arn = db_cluster_parameter_group["DBClusterParameterGroupArn"] # get tags tags = await hub.tool.aws.neptune.tag.get_tags_for_resource( ctx, resource_arn=arn ) if not tags["result"]: result["result"] = False result["comment"].append(tags["comment"]) return result tags = tags["ret"] # get parameters parameters_raw = ( await hub.exec.boto3.client.neptune.describe_db_cluster_parameters( ctx=ctx, DBClusterParameterGroupName=db_cluster_parameter_group_name ) ) if not parameters_raw["result"]: result["result"] = False result["comment"].append(parameters_raw["comment"]) return result parameters_raw = parameters_raw.get("ret") parameters = parameters_raw.get("Parameters") result["ret"].append( hub.tool.aws.neptune.conversion_utils.convert_raw_db_cluster_parameter_group_to_present( raw_resource=db_cluster_parameter_group, idem_resource_name=db_cluster_parameter_group_name, tags=tags, parameters=parameters, ) ) return result