title: Airflow: how to use trigger parameters in functions
body: <p>We are using Airflow's <code>KubernetesPodOperator</code> for our data pipelines. What we would like to add is the option to pass in parameters via the UI.</p>
  <p>We currently use it in a way that we have different yaml files that are storing the parameters for the operator, and instead of calling the operator directly we are calling a function that does some prep and returns the operator like this:</p>
  <pre><code>def prep_kubernetes_pod_operator(yaml):
  
      # ... read yaml and extract params
  
      return KubernetesPodOperator(params)
  
  with DAG(...):
      
      task1 = prep_kubernetes_pod_operator(yaml)
  
  </code></pre>
  <p>For us this works well and we can keep our dag files pretty lightweight, however now we would like to add the functionality that we can add some extra params via the UI. I understand that the trigger params can be accessed via <code>kwargs['dag_run'].conf</code>, but I had no success pulling these into the Python function.</p>
  <p>Another thing I tried is to create a custom operator because that recognises the args, but I couldn't manage to call the <code>KubernetesPodOperator</code> in the execute part (and I guess calling an operator in an operator is not right solution anyways).</p>
  <p><strong>Update:</strong></p>
  <p>Following NicoE's advice, I started to extend the <code>KubernetesPodOperator</code> instead.</p>
  <p>The error I am having now is that when I am parsing the yaml and assign the arguments after, the parent arguments become tuples and that throws a type error.</p>
  <p>dag:</p>
  <pre><code>task = NewKPO(
      task_id=&quot;task1&quot;,
      yaml_path=yaml_path)
  </code></pre>
  <p>operator:</p>
  <pre><code>class NewKPO(KubernetesPodOperator):
     @apply_defaults
     def __init__(
             self,
             yaml_path: str,
             name: str = &quot;default&quot;,
             *args,
             **kwargs) -&gt; None:
         self.yaml_path = yaml_path
         self.name = name
         super(NewKPO, self).__init__(
             name=name, # DAG is not parsed without this line - 'key has to be string'
             *args,
             **kwargs)
  
     def execute(self, context):
         # parsing yaml and adding context[&quot;dag_run&quot;].conf (...)
         self.name = yaml.name
         self.image = yaml.image
         self.secrets = yaml.secrets
         #(...) if i run a type(self.secrets) here I will get tuple
         return super(NewKPO, self).execute(context)
  </code></pre>