<p>I am working on Airflow 1.10.</p>
<p>I have problem with running commands on KubernetesPodOperator, where entire command is evaluated during DAG runtime.</p>
<p>I am generating command in DAG runtime as some of command's arguments depends on parameters passed by user.</p>
<p>As I read from documentation
KubernetesPodOperator expects list of strings or list of jinja templates:</p>
<pre><code>    :param arguments: arguments of the entrypoint. (templated)
        The docker image's CMD is used if this is not provided.
</code></pre>
<p>I have PythonOperator which generates command and push it to XCOM and KubernetesPodOperator
where in arguments I pass command generated by PythonOperator.</p>
<pre><code>from airflow.operators.python_operator import PythonOperator
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator


def command_maker():
    import random # random is to illustrate that we don't know arguments value before runtime
    return f&quot;my_command {random.randint(1, 10)} --option {random.randint(1, 4)}&quot;

def create_tasks(dag):
    first = PythonOperator(
        task_id=&quot;generate_command&quot;,
        python_callable=command_maker,
        provide_context=True,
        dag=dag,
    )
    second = KubernetesPodOperator(
        namespace='some_namespace',
        image='some_image',
        name='execute_command',
        dag=dag,
        arguments=[f'{{ ti.xcom_pull(dag_id=&quot;{dag.dag_id}&quot;, task_ids=&quot;generate_command&quot;, key=&quot;return_value&quot;)}}']
    )
    second.set_upstream(first)
</code></pre>
<p>Unfortunately KubernetesPodOperator doesn't run this command correctly as he tries to run something like this:</p>
<pre><code>[my_command 4 --option 2]
</code></pre>
<p>Is there way to eval this list at KubernetesPodOperator runtime
or am I enforced to push all runtime arguments into separate XCOM's?
I would like to avoid such solution as it required lot of changes in my project.</p>
<pre><code>         arguments=[
            &quot;my_command&quot;,
            f'{{ ti.xcom_pull(dag_id=&quot;{dag.dag_id}&quot;, task_ids=&quot;generate_command&quot;, key=&quot;first_argument&quot;)}}',
            &quot;--option&quot;,
            f'{{ ti.xcom_pull(dag_id=&quot;{dag.dag_id}&quot;, task_ids=&quot;generate_command&quot;, key=&quot;second_argument&quot;)}}',
         ]
</code></pre>

domandinho
<p>To have working solution for Airflow 1.10 I had to use BaseOperator.pre_execute hook.</p>
<pre><code>from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.lineage import prepare_lineage

class UnpackCommandKubernetesPodOperator(KubernetesPodOperator):
    @prepare_lineage
    def pre_execute(self, context):
        self.arguments = self.arguments[0].split(&quot; &quot;)
</code></pre>
