VhQLdZdZdZddlZddlZddlZ ddlZddlm Z ddl m Z ddl m Z ddlmZd d gZgd Zd Zd ZdZGddeZGddeZdZdZdZdZdZdZdZdZ dZ!dZ"dZ#dZ$dZ%d Z&d!Z'd"Z(d#Z)d$Z*d%Z+e,d&k(re+yy#e$rYwxYw)'aL --- module: data_pipeline version_added: 1.0.0 author: - Raghu Udiyar (@raags) - Sloane Hertel (@s-hertel) short_description: Create and manage AWS Datapipelines description: - Create and manage AWS Datapipelines. Creation is not idempotent in AWS, so the C(uniqueId) is created by hashing the options (minus objects) given to the datapipeline. - The pipeline definition must be in the format given here U(https://docs.aws.amazon.com/datapipeline/latest/APIReference/API_PutPipelineDefinition.html#API_PutPipelineDefinition_RequestSyntax). - Operations will wait for a configurable amount of time to ensure the pipeline is in the requested state. options: name: description: - The name of the Datapipeline to create/modify/delete. required: true type: str description: description: - An optional description for the pipeline being created. default: '' type: str objects: type: list elements: dict default: [] description: - A list of pipeline object definitions, each of which is a dict that takes the keys I(id), I(name) and I(fields). suboptions: id: description: - The ID of the object. type: str name: description: - The name of the object. type: str fields: description: - Key-value pairs that define the properties of the object. - The value is specified as a reference to another object I(refValue) or as a string value I(stringValue) but not as both. type: list elements: dict suboptions: key: type: str description: - The field identifier. stringValue: type: str description: - The field value. - Exactly one of I(stringValue) and I(refValue) may be specified. refValue: type: str description: - The field value, expressed as the identifier of another object. - Exactly one of I(stringValue) and I(refValue) may be specified. parameters: description: - A list of parameter objects (dicts) in the pipeline definition. type: list elements: dict default: [] suboptions: id: description: - The ID of the parameter object. attributes: description: - A list of attributes (dicts) of the parameter object. type: list elements: dict suboptions: key: description: The field identifier. type: str stringValue: description: The field value. type: str values: description: - A list of parameter values (dicts) in the pipeline definition. type: list elements: dict default: [] suboptions: id: description: The ID of the parameter value type: str stringValue: description: The field value type: str timeout: description: - Time in seconds to wait for the pipeline to transition to the requested state, fail otherwise. default: 300 type: int state: description: - The requested state of the pipeline. choices: ['present', 'absent', 'active', 'inactive'] default: present type: str tags: description: - A dict of key:value pair(s) to add to the pipeline. type: dict default: {} aliases: ['resource_tags'] extends_documentation_fragment: - amazon.aws.common.modules - amazon.aws.region.modules - amazon.aws.boto3 a # Note: These examples do not set authentication details, see the AWS Guide for details. # Create pipeline - community.aws.data_pipeline: name: test-dp region: us-west-2 objects: "{{pipelineObjects}}" parameters: "{{pipelineParameters}}" values: "{{pipelineValues}}" tags: key1: val1 key2: val2 state: present # Example populating and activating a pipeline that demonstrates two ways of providing pipeline objects - community.aws.data_pipeline: name: test-dp objects: - id: "DefaultSchedule" name: "Every 1 day" fields: - "key": "period" "stringValue": "1 days" - "key": "type" "stringValue": "Schedule" - "key": "startAt" "stringValue": "FIRST_ACTIVATION_DATE_TIME" - id: "Default" name: "Default" fields: - "key": "resourceRole" "stringValue": "my_resource_role" - "key": "role" "stringValue": "DataPipelineDefaultRole" - "key": "pipelineLogUri" "stringValue": "s3://my_s3_log.txt" - "key": "scheduleType" "stringValue": "cron" - "key": "schedule" "refValue": "DefaultSchedule" - "key": "failureAndRerunMode" "stringValue": "CASCADE" state: active # Activate pipeline - community.aws.data_pipeline: name: test-dp region: us-west-2 state: active # Delete pipeline - community.aws.data_pipeline: name: test-dp region: us-west-2 state: absent a changed: description: whether the data pipeline has been modified type: bool returned: always sample: changed: true result: description: - Contains the data pipeline data (data_pipeline) and a return message (msg). If the data pipeline exists data_pipeline will contain the keys description, name, pipeline_id, state, tags, and unique_id. If the data pipeline does not exist then data_pipeline will be an empty dict. The msg describes the status of the operation. returned: always type: dict N)to_text)camel_dict_to_snake_dict)is_boto3_error_code)AnsibleCommunityAWSModuleACTIVE SCHEDULED)INACTIVEPENDINGFINISHEDDELETING ACTIVATING DEACTIVATINGz(^.*Pipeline with id: {0} does not exist$c eZdZy)DataPipelineNotFoundN__name__ __module__ __qualname__o/home/dcms/DCMS/lib/python3.12/site-packages/ansible_collections/community/aws/plugins/modules/data_pipeline.pyrrrrc eZdZy)TimeOutExceptionNrrrrrrrrrc`|j}|dD]}|d|k(s |dcSt)zReturn pipeline id for the given pipeline name :param object client: boto3 datapipeline client :param string name: pipeline name :returns: pipeline id :raises: DataPipelineNotFound pipelineIdListnameid)list_pipelinesr)clientr pipelinesdps r pipeline_idr#sD%%'I() f: d8O rc^ |j|gS#tddg$rtwxYw)zReturn pipeline description list :param object client: boto3 datapipeline client :returns: pipeline description dictionary :raises: DataPipelineNotFound ) pipelineIdsPipelineNotFoundExceptionPipelineDeletedException)describe_pipelinesrrr dp_ids rpipeline_descriptionr+s>#((eW(==  ;=WX Y#""#s,cvt||}|dddD]}|d|k(s |dcStd|d)a6Return a pipeline field from the pipeline description. The available fields are listed in describe_pipelines output. :param object client: boto3 datapipeline client :param string dp_id: pipeline id :param string field: pipeline description field :returns: pipeline field information pipelineDescriptionListrfieldskey stringValuez Field key z not found!)r+KeyError)r r*fielddp_description field_keys rpipeline_fieldr5s]*&%8N#$=>qA(K, U u $]+ +, Zwk2 33rcrt|dzD]!}||i|rytjd#t)aRun func with the provided args and kwargs, and wait until timeout for truthy return value :param int timeout: time to wait for status :param function func: function to run, should return True or False :param args func_args: function args to pass to func :param kwargs func_kwargs: function key word args :returns: True if func returns truthy within timeout :raises: TimeOutException T)rangetimesleepr)timeoutfunc func_args func_kwargscounts rrun_with_timeoutr@sAw"}%  *k * JJrN  rcL tt||S#t$rYywxYw)zCheck if datapipeline exists :param object client: boto3 datapipeline client :param string dp_id: pipeline id :returns: True or False F)boolr+rr)s rcheck_dp_existsrC1s,(788 s  ##cht|ts ttt ||d|vS)zChecks if datapipeline matches states in status list :param object client: boto3 datapipeline client :param string dp_id: pipeline id :param list status: list of states to check against :returns: True or False @pipelineStater2) isinstancelistAssertionErrorrBr5)r r*statuss rcheck_dp_statusrK@s1 fd # vu4DEO PPrc.|||f}t|tg|SN)r@rK)r r*rJr;argss rpipeline_status_timeoutrONs E6 "D G_ B? C%$C%(4DDc|jjd}|jjd} t||}t |dt vrd}n(|j|  t||t | d }t||}|d|dd}||fS#t$r|j d|dYywxYw#t$r|j d|d |d YdwxYw)zDeactivates pipelinerr;rSrTrUrErFFrWrZz% failed to deactivate within timeout r[Tz deactivated.r\) r^r_r#rr`r5DP_INACTIVE_STATESdeactivate_pipelinerOrrcrds rrkrk~smm'Gmm *GCFG,fe+;<@RR""e"4  #FE:LV] ^ vu-M&y 6F V + C~gYjABC     $WI-RSZR[[cd   s# B.CB>=B>"C&%C&cp|j| t||tg|y#t$rYywxYw)NrW)r r*rJr;T)delete_pipelinerOPIPELINE_DOESNT_EXISTr)r*r r;s r_delete_dp_with_checkros= e,vUDYCZdkl s ) 55c(|jjd}|jjd} t||}t|||d}id|d d }|fS#t$rd}Yt $r|j d|d|dY=wxYw) zDeletes pipelinerr;TFrSz! failed to delete within timeout r[rUz deletedr\)r^r_r#rorrr`)r rerfr;r*rgrhs rrmrmsmm'Gmm *G  FG,eVW5y1F V      )J7)S[\   sA B-!BBc t|j}dDcgc]}|j|dc}tj|dj d}t j|j}|Scc}w)N)objectsr;T) sort_keyszutf-8) dictr^popjsondumpsencodehashlibmd5 hexdigest)redataeach json_data hashed_datas rbuild_unique_idrsf  D&<=dTXXdD= 44077@I++i(224K >sBcj|jDcgc]\}}t||c}}Scc}}w)zReformats tags :param dict tags: dict of data pipeline tags (e.g. {key1: val1, key2: val2, key3: val3}) :returns: list of dicts (e.g. [{key: key1, value: val1}, {key: key2, value: val2}, {key: key3, value: val3}]) )r/value)itemsrt)tagskvs r format_tagsrs).2ZZ\ :TQDQa :: :s/ct||dd}t||d|d<t||d|d<|d=t|}|S) zGet the current state of the data pipeline and reformat it to snake_case for exit_json :param object client: boto3 datapipeline client :param string dp_id: pipeline id :returns: reformatted dict of pipeline description r-runiqueIdrF unique_idrEpipeline_stater.)r+r5r)r r*r"s rrcrcs^ fe ,-F G JB%VU*EB{O)&%?OPB 8 "" %B Irc2i}d}d}t|} t||}tt||d} | |k7rd}d}nC|j |d} | |k7rt ||||\}} nd|d } t ||} | | d }|||fS#t$rd}YwxYw) zmCheck if there's another pipeline with the same unique_id and if so, checks if the object needs to be updatedFrrF NEW_VERSIONTrWpipelineObjectsrSz is presentr\)rr#rr5get_pipeline_definitiondefine_pipelinercr) r rerrrrfrhrg create_dpr* dp_unique_id dp_objectsrVr]s r diff_pipeliners FGI 'IFG,~fe:NO 9 $#GI 7757IJ[\JW$.vvwN 'wi{;&vu5M!.F gv %%  sA0B BBc|jjd}t||ddk(r d|d}d}||fS|rX|jjd}|jjd } |j|||| d|d }d }||fSd}d}||fS#tj j tj jf$r$} |j| d |dYd} ~ fSd} ~ wwxYw)zPuts pipeline definitionrrErFr rSz1 is unable to be updated while in state FINISHED.F parametersvalues)rXrparameterObjectsparameterValuesz has been updated.Tz*Failed to put the definition for pipeline z. Check that string/reference fieldsare not empty and that the number of objects in the pipeline does not exceed maximum allowedobjectsrUN) r^r_r5put_pipeline_definitionbotocore exceptions ClientError BotoCoreError fail_json_aws) r rerrr*rfrVrgrres rrr s7mm'Gfe+;< Jwi'XY2 C</ ]]&&|4 ""8,   * * 'Jhn + #7)+=>CG C< C<##//1D1D1R1RS   @ J !   C< s2B7C6C11C6c<|jjd}|jjdd}|jjdd}|jjd}|jjd}t|}t|||||\}} } | dk(rt ||\} } |r^ t |}|j |||| } | d } t|| |t||| \} }d} t!|| }|d |d|zd} | | fS#t$r|jd |d |d YStjjtjjf$r!}|j|d|dYd}~d}~wwxYw)z;Creates datapipeline. Uses uniqueId to achieve idempotency.rrrN descriptionrrr;r)rrrrrXrSz! failed to create within timeout r[rUz#Failed to create the data pipeline .Tz created.r\)r^r_rrrmrcreate_pipelinerQrr`rrrrrrrc)r rerfrrrrr;rrrgrhcreation_resultr"r*rrVr]s rrr.smm'Gmm 40G--##M26K ==  V $Dmm *G'I!.vvw SZ![Iw-#266#B  t$D''WyVahl'mB|$E #FE7 ;'vvwF "651 *#G9I6<  V '     $WI-NwiW_`  ##//1D1D1R1RS   9'!D !   s?2D "F6F:FFcttdtddtddgdtddgdtdd d td gd tddidgtddgd}t|d} |jd}|jjd}|d k(rt|\}}n>|dk(rt|\}}n)|dk(rt|\}}n|dk(rt|\}}|jy#tjj tjj f$r}|j|dYd}~d}~wwxYw)NT)requiredFr)rdefaultrHrt)rtyperelementsinti,)rrrpresent)rabsentactiveinactive)rchoices resource_tags)rrraliases)rrrrrr;staterr) argument_specsupports_check_mode datapipelinezFailed to connect to AWSrUrrrr)rhrg)rtAnsibleAWSModuler rrrrrr^r_rrmrbrk exit_json)rrer rrrgrhs rmainr[sc 4 %4e&"vNVR&Qe%=9.YZ 5vrOCT UUfM MMu UF@~. MM  g &E )&&9 ( )&&9 ( +FF; * -ff= FG4    + +X-@-@-N-N O@Q$>??@sD7E,E''E,__main__)- DOCUMENTATIONEXAMPLESRETURNryrvr9r ImportErroransible.module_utils._textr0ansible.module_utils.common.dict_transformationsransible_collections.community.aws.plugins.module_utils.modulesrrrarjDP_ACTIVATING_STATEDP_DEACTIVATING_STATErn Exceptionrrr#r+r5r@rCrKrOrQrbrkrormrrrcrrrrrrrrrsw r8 t "  /U\xk*D"&B 9  y   #4$.  Q= = #L>.;4 &FD*Z5> zFS   sBB#"B#