
    M                         S r SSKrSSKrSSKrSSKrSSKJr  SSKJr	  SSK
Jr  SSKJr  SSKJr  SSKJr  SSKJr  SS	KJr  SS
KJr  SrSrSrSrSrSrSrSr\R<                   " S S\R>                  5      5       r g)z<Command to run an Airflow CLI sub-command in an environment.    N)environments_util)util)base)image_versions_util)resource_args)execution_utils)log)
console_iozairflow-workera  Because Cloud Composer manages the Airflow metadata database for your environment, support for the Airflow `{}` subcommand is being deprecated. To avoid issues related to Airflow metadata, we recommend that you do not use this subcommand unless you understand the outcome.   
      g      ?g      ?c                       \ rS rSrSr\R                  r\S 5       rS r	S r
S rS rS rS	 rS
 rS rS rS rS rSrg)Run0   a  Run an Airflow sub-command remotely in a Cloud Composer environment.

Executes an Airflow CLI sub-command remotely in an environment. If the
sub-command takes flags, separate the environment name from the sub-command
and its flags with ``--''. This command waits for the sub-command to
complete; its exit code will match the sub-command's exit code.

Note: Airflow CLI sub-command syntax differs between Airflow 1 and Airflow 2.
Refer to the Airflow CLI reference documentation for more details.

## EXAMPLES

  The following command in environments with Airflow 2:

  {command} myenv dags trigger -- some_dag --run_id=foo

is equivalent to running the following command from a shell inside the
*my-environment* environment:

  airflow dags trigger --run_id=foo some_dag

The same command, but for environments with Airflow 1.10.14+:

  {command} myenv trigger_dag -- some_dag --run_id=foo

is equivalent to running the following command from a shell inside the
*my-environment* environment:

  airflow trigger_dag some_dag --run_id=foo

The following command (for environments with Airflow 1.10.14+):

  {command} myenv dags list

is equivalent to running the following command from a shell inside the
*my-environment* environment:

  airflow dags list
c                    [         R                  " US5        SnUR                  SS[        U R                  R                  5       5      SR                  SR                  [        U R                  R                  5        VVs/ s H=  u  p4SR                  X4R                  =(       d    SUR                  =(       d    S5      PM?     snn5      5      U5      S	9  / nU R                  R                  5        Hi  u  pdUR                  (       d  M  UR                  S
R                  USR                  [        UR                  R                  5       5      5      5      5        Mk     UR                  S5        UR                  SS[        R                  SR                  SR                  U5      5      S9  UR                  SS[        R                   SSS9  g s  snnf )Nz"in which to run an Airflow commandzThttps://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html
subcommand
SUBCOMMANDzThe Airflow CLI subcommand to run. Available subcommands include (listed with Airflow versions that support): {} (see {} for more info)., z{} [{}, {})z**)metavarchoiceshelpz- {}: {}z;- all other subcommands: all nested subcommands are allowedsubcommand_nestedSUBCOMMAND_NESTEDzeAdditional subcommand in case it is nested. The following is a list of allowed nested subcommands:
{}
)r   nargsr   cmd_argsCMD_ARGSz)Command line arguments to the subcommand.z4{command} myenv trigger_dag -- some_dag --run_id=foo)r   r   r   example)r   AddEnvironmentResourceArgadd_argumentlistSUBCOMMAND_ALLOWLISTkeysformatjoinsorteditemsfrom_version
to_versionallowed_nested_subcommandsappendargparseOPTIONAL	REMAINDER)clsparserdoc_urlcmdrallowed_nested_subcommands_helpsub_cmds          (lib/surface/composer/environments/run.pyArgsRun.Args\   s   ++46 eG
S--2245()/)) ),(@(@(F(F(H )Ifc (..sNN4Jd/0||/CtE(H 
 #*$  %  ')#..446
 ))%,,


iiq;;@@BCD 7 $**E # &:;
<  	   8F  HCs   AGc                 :  ^^ SSSSSSS.mU R                  U5      nUU4S jnU" U5      (       ak  [        UR                  =(       d    / 5      R                  SS15      (       a6  UR                  =(       d    / Ul        UR                  R	                  S5        ggg)a  Bypasses confirmations with "yes" responses.

Prevents certain Airflow CLI subcommands from presenting a confirmation
prompting (which can make the gcloud CLI stop responding). When necessary,
bypass confirmations with a "yes" response.

Args:
  args: argparse.Namespace, An object that contains the values for the
    arguments specified in the .Args() method.
  airflow_version: String, an Airflow semantic version.
z1.10.6N)backfill
delete_dag)dagsr:   )r<   delete)tasksclear)dbcleanc                    > U T;   a  OU S   T;   a  U S   n OgTU    S L =(       d    [         R                  " TTU    5      S:  $ )Nr   F)image_versions_command_utilCompareVersions)sairflow_versionprompting_subcommandss    r6   _IsPromptingSubcommand<Run.BypassConfirmationPrompt.<locals>._IsPromptingSubcommand   sc    	
#	#Q4((aD#A&$. B)99!#8#;=@ABC    z-yz--yes)_GetSubcommandTwoLevelsetr   
isdisjointr+   )selfargsrF   subcommand_two_levelrH   rG   s     `  @r6   BypassConfirmationPromptRun.BypassConfirmationPrompt   s     "    66t<
C 	344DMMR ++T7O<<mm)rdm
mm7# 	= 	5rJ   c           	      X   S/ SQ/0nS nU R                  U5      nUR                  U/ 5       Hy  n[        U5      R                  [        UR                  =(       d    / 5      5      (       d  M@  [
        R                  " SR                  SR                  U5      U" U5      5      5      e   g)zPrevents running Airflow CLI commands without required arguments.

Args:
  args: argparse.Namespace, An object that contains the values for the
    arguments specified in the .Args() method.
)userscreate)z-pz
--passwordz--use-random-passwordc                     U  Vs/ s H  nSR                  U5      PM     nnSR                  SR                  U5      5      $ s  snf )Nz"{}"z[{}]r   )r$   r%   )r   aquoted_argss      r6   _StringifyRequiredCmdArgs>Run.CheckForRequiredCmdArgs.<locals>._StringifyRequiredCmdArgs   s<    /78x!V]]1%xk8]]499[122 9s   AzMThe subcommand "{}" requires one of the following command line arguments: {}. N)	rK   getrL   rM   r   command_utilErrorr$   r%   )rN   rO   required_cmd_argsrY   rP   subcommand_required_cmd_argss         r6   CheckForRequiredCmdArgsRun.CheckForRequiredCmdArgs   s     	KL3  66t<
 ):(=(=b)"$	)	*	5	5c$--:M26N	O	O  #V-.)*FGIJ 	J)"rJ   c                     SnUR                   [        R                  ;   a3  [        R                  " [
        R                  UR                   5      SSS9nU$ )NTF)messagedefaultcancel_on_no)r   r]   SUBCOMMAND_DEPRECATIONr
   PromptContinueDEPRECATION_WARNINGr$   )rN   rO   responses      r6   DeprecationWarningPromptRun.DeprecationWarningPrompt   sH    H,===**%,,T__=h OrJ   c                     UR                   S4nUR                  (       a  UR                   UR                  4nU$ UR                  (       a  UR                   UR                  S   4nU$ )a  Extract and return two level nested Airflow subcommand in unified shape.

There are two ways to execute nested Airflow subcommands via gcloud, e.g.:
1. {command} myenv users create -- -u User
2. {command} myenv users -- create -u User
The method returns here (users, create) in both cases.

It is possible that first element of args.cmd_args will not be a nested
subcommand, but that is ok as it will not break entire logic.
So, essentially there can be subcommand_two_level = ['info', '--anonymize'].

Args:
  args: argparse.Namespace, An object that contains the values for the
    arguments specified in the .Args() method.

Returns:
  subcommand_two_level: two level subcommand in unified format
Nr   )r   r   r   )rN   rO   rP   s      r6   rK   Run._GetSubcommandTwoLevel   s`    ( !OOT2"oot/E/EF   
"oot}}Q/?@rJ   c                 t  ^ U4S jnS mU R                  U5      u  pEU" XBU R                  UR                     R                  U R                  UR                     R                  5        U R                  UR                     R
                  (       d  g SR                  XE5      nXPR                  UR                     R
                  ;   ag  U" XbU R                  UR                     R
                  U   R                  U R                  UR                     R
                  U   R                  5        g T" Xb5        g )Nc                 P   > [         R                  " XU5      (       d	  T" X5        g g N)rC   IsVersionInRange)commandrF   r(   r)   _RaiseLackOfSupportErrors       r6   _CheckIsSupportedSubcommandFRun.CheckSubcommandAirflowSupport.<locals>._CheckIsSupportedSubcommand	  s(    (99
5 5 :5rJ   c                 L    [         R                  " SR                  X5      5      e)NzWThe subcommand "{}" is not supported for Composer environments with Airflow version {}.)r]   r^   r$   )rs   rF   s     r6   rt   CRun.CheckSubcommandAirflowSupport.<locals>._RaiseLackOfSupportError  s&    &&,fW&FI IrJ   z{} {})rK   r"   r   r(   r)   r*   r$   )rN   rO   rF   ru   r   r   two_level_subcommand_stringrt   s          @r6   CheckSubcommandAirflowSupport!Run.CheckSubcommandAirflowSupport  s   ;I
 %)$?$?$E!J!!$//2??!!$//2==?
 $$334").."O55334!
%

#
#DOO
4%%&799E

#
#DOO
4%%&799CE :LrJ   c                     UR                   (       a4  [        R                  " USS 5      (       d  [        R                  " S5      eg g )Nz1.10.14zgNested subcommands are supported only for Composer environments with Airflow version 1.10.14 or higher.)r   rC   rr   r]   r^   )rN   rO   rF   s      r6   #CheckSubcommandNestedAirflowSupport'Run.CheckSubcommandNestedAirflowSupport+  sG    '88Y. .45 5. 	rJ   c                     UR                   R                  =(       a     UR                   R                  R                  nU(       a"  [        R                  " [        U5      S-   5      $ U$ )Nz Make sure you have followed https://cloud.google.com/composer/docs/how-to/accessing/airflow-cli#private-ip to enable access to your private Cloud Composer environment from your machine.)configprivateEnvironmentConfigenablePrivateEnvironmentr]   r^   str)rN   errorenv_obj
is_privates       r6   ConvertKubectlErrorRun.ConvertKubectlError3  s]    // 	I//HH  
e*  LrJ   c                 6    [         R                  " SU5      S   $ )Nz-airflow-([\d\.]+)r   )refindall)rN   image_versions     r6   _ExtractAirflowVersionRun._ExtractAirflowVersionA  s    ::+];A>>rJ   c                 \   UR                   R                  n[        R                  " U5      nSU;  n[        R                  " XCS5          UR                   R
                  R                  nU R                  U5      nU R                  X5        U R                  X5        [        R                  " U5      n[        R                  " [        US9n	[        R                  R                  SR!                  U5      5        U R#                  X5        SU	S/n
U(       a  U
R%                  S5        U
R'                  S[(        S	S
UR*                  /5        UR,                  (       a  U
R%                  UR,                  5        UR.                  (       a  U
R'                  UR.                  5        [        R0                  " [        R2                  " X5      [        R4                  R                  S9   SSS5        g! [        R6                   a  nU R9                  X5      eSnAff = f! , (       d  f       g= f)as  Runs Airflow command using kubectl on the GKE Cluster.

This mode the command is executed by connecting to the cluster and
running `kubectl pod exec` command.
It requires access to GKE Control plane.

Args:
  args: argparse.Namespace, An object that contains the values for the
    arguments specified in the .Args() method.
  env_obj: Cloud Composer Environment object.
zno-ttyN)
pod_substrkubectl_namespacez?Executing within the following Kubernetes cluster namespace: {}execz--stdinz--ttyz--containerz--airflow)out_func)r   
gkeClusterr]   ExtractGkeClusterLocationIdTemporaryKubeconfigsoftwareConfigimageVersionr   rz   r}   FetchKubectlNamespace	GetGkePodWORKER_POD_SUBSTRr	   statusPrintr$   rQ   r+   extendWORKER_CONTAINERr   r   r   RunKubectlCommandAddKubectlNamespaceoutKubectlErrorr   )rN   rO   r   
cluster_idcluster_location_idttyr   rF   
kubectl_nspodkubectl_argses               r6   _RunKubectlRun._RunKubectlD  s    **J&BB7K
$
C		)	)
355BB55mD**4A00G!77F
$$(JH 	


#	% 	%%d<Y/


g
&,dItO	Q!!


d44
5==


dmm
,&&,,ZFWW]]	$9
 
> && 3&&q223?
 
s+   	HFG00HHHH
H+c           
         UR                   R                  R                  nU R                  U5      nUR                  R
                  R                  5       nU R                  X5        U R                  X5        U R                  X5        UR                  /nUR                  (       a  UR                  UR                  5        UR                  (       a  UR                  UR                  5        [        R                   R#                  SR%                  SR'                  U5      5      5        [(        R*                  " UR                  UR                  =(       d    SUR                  =(       d    / UU R-                  5       S9nU(       aI  UR.                  (       a8  [        R                   R#                  SR%                  UR.                  5      5        UR.                  (       d  [0        R2                  " S5      e[        R                   R#                  S5        SnS	n	S
n
[4        nS nSnSnU(       Gds  U(       Gdk  S n [6        R8                  " 5          [:        R<                  " U[>        R@                  " [B        * [B        5      -   5        [(        RD                  " UR.                  URF                  URH                  U	UU R-                  5       S9nS
n
URJ                  nURL                  nURO                  S S9  S S S 5        U
[X        :X  a  [0        R2                  " S5      eU(       d  [[        U[\        -  [^        5      nOT[4        nU H6  n[        R"                  " UR`                  (       a  UR`                  OS5        M8     US   Rb                  S	-   n	U(       d
  U(       d  GMk  U(       a  URd                  (       a  URd                  Rf                  (       a  URd                  Rh                  (       a9  [        Rh                  " SR%                  URd                  Rh                  5      5        [        Rh                  " SR%                  URd                  Rf                  5      5        [k        URd                  Rf                  5        g g g g ! , (       d  f       GN= f! [P         Ga    [        R                   R#                  S5         [        RR                  " S5        [(        RT                  " UR.                  URF                  UURH                  UU R-                  5       S9n[        RR                  " S[W        U5      -   5        U(       a:  URL                  (       a)  URL                   H  n[        R"                  " U5        M     U(       a  SnSn GN!   [        RR                  " S5        U
S	-  n
  GN= f  U
S	-  n
 GN= f)Nz(Executing the command: [ airflow {} ]...r[    )rs   r   
parametersenvironment_refrelease_trackz)Command has been started. execution_id={}zBCannot execute subcommand for environment. Got empty execution Id.z#Use ctrl-c to interrupt the commandF   r   )execution_idpod_namepod_namespacenext_line_numberr   r   c                     U R                   $ rq   )
lineNumber)lines    r6   <lambda>Run._RunApi.<locals>.<lambda>  s    doorJ   )keyzInterrupting the command...zStopping the airflow command...)r   r   forcer   r   r   zStop airflow command result...Tz7Error during stopping airflow command. Retrying pollingz$Cannot fetch airflow command status.zError message: {}zCommand exit code: {})6r   r   r   r   CONCEPTSenvironmentParserz   r}   rQ   r   r   r+   r   r   r	   r   r   r$   r%   environments_api_utilExecuteAirflowCommandReleaseTrackexecutionIdr]   r^   DEFAULT_POLL_TIME_SECONDSr   RaisesKeyboardInterrupttimesleeprandomuniformPOLL_JITTER_SECONDSPollAirflowCommandr   podNamespace	outputEndoutputsortKeyboardInterruptdebugStopAirflowCommandr   MAX_CONSECUTIVE_POLL_ERRORSminEXP_BACKOFF_MULTIPLIERMAX_POLL_TIME_SECONDScontentr   exitInfoexitCoder   exit)rN   rO   r   r   rF   env_refr2   execute_result
output_end	next_linecur_consequetive_poll_errorswait_time_secondspoll_resultinterrupted
force_stoplinesstop_resultr   s                     r6   _RunApiRun._RunApiw  sZ   NN11>>M11-@Omm''--/G&&t=,,TC!!$8??
C	jj''(}}	jjJJ299#((3-H +@@))/R==&B'')N .44	jj
5
<
<(( %%
N  JJ:;JI#$ 1KKJe)*446
** 335HIJ .@@)55%))*77(% --/+ *+
&",,*$$%
**5*
6! 7T 
&)D	D  !GHH 668M
 6D
))DLLDLLb
9 "I((1,	q t {++0D0D0M0M				#	#		%,,[-A-A-G-GHI	ii'..{/C/C/L/LMN
;(()	 1N+{o 76"  ,

67	,
))5
6-@@)55%))*77% --/+ ))4S5EE
F[//#**iio +J+	,
))M
N
&!
+
&*$)$sD   <R B'R8R 
RR R *V=C	VV1-V=1	V=c                    U R                  U5        U R                  U5        [        R                  " U R	                  5       S9R
                  R                  R                  nUR                  R                  R                  5       n[        R                  " X0R	                  5       S9nUR                  U:w  a/  [        R                  " SR!                  UR                  5      5      e["        R$                  " UR&                  R(                  R*                  S9(       a  U R-                  X5        g U R/                  X5        g )N)r   zGCannot execute subcommand for environment in state {}. Must be RUNNING.)r   )rk   ra   api_utilGetMessagesModuler   EnvironmentStateValueValuesEnumRUNNINGr   r   r   r   Getstater]   r^   r$   rC   %IsVersionAirflowCommandsApiCompatibler   r   r   r   r   )rN   rO   running_stater   r   s        r6   r   Run.Run  s    !!$'  &..'')k&&ww  mm''--/G#''002G }}%#VGMM2  #HHnn33@@ ll4!
t%rJ    N)__name__
__module____qualname____firstlineno____doc__r]   r"   classmethodr7   rQ   ra   rk   rK   rz   r}   r   r   r   r   r   __static_attributes__r   rJ   r6   r   r   0   sj    &P &::3H 3Hj,$\JB :"MH5?13fi*V&rJ   r   )!r   r,   r   r   r   googlecloudsdk.api_lib.composerr   r   r   r   googlecloudsdk.callioper   #googlecloudsdk.command_lib.composerr   rC   r   r]   googlecloudsdk.corer   r	   googlecloudsdk.core.consoler
   r   r   ri   r   r   r   r   r   DefaultUniverseOnlyCommandr   r   rJ   r6   <module>r     s    C   	  V < ( b = D / # 2$ # 5         H&$,, H& H&rJ   