Jinja template not rendering in a custom sensor that internally uses a custom hook

[sid-habu]

@task(multiple_outputs=True)
def get_data_plane_conf(**kwargs) -> dict:
    return {
        "connection_name": "test",
        "connection_region" : "test-region"
    }

data_plane_conf = get_data_plane_conf()

with TaskGroup(group_id='group_1') as group1:
    t1 = CustomOperator(
        task_id='task1',
        conn_id=f'{data_plane_conf["connection_name"]}',
        do_xcom_push=True,
    )
    
    t2 = CustomSensor(
        task_id='task2',
        conn_id=f'{data_plane_conf["connection_name"]}',
        app_name="{{ task_instance.xcom_pull(task_ids='group_1.task1')['metadata']['name'] }}",
    )

    t1 >> t2

class CustomOperator(BaseOperator):
  template_fields: Sequence[str] = ["conn_id"]

class CustomSensor(SomeBaseSensor):
  template_fields: Sequence[str] = ["conn_id", "app_name"]

  def __init__(
        self,
        *,
        conn_id: str,
        app_name: str,
        **kwargs,
  ) -> None:
      super().__init__(
                       conn_id=conn_id,
                       **kwargs)
      self.conn_id = conn_id
      self.app_name = app_name
      self.hook = CustomHook(connection_name=conn_id)

I have a custom Operator and a Sensor that define conn_id in the template_field. I have an upstream PythonOperator that returns an XCOM. Only the first task (t1) gets the templated value test from the XCOM within the Task Group. conn_id in t2 shows up as {{ task_instance.xcom_pull(task_ids='get_data_plane_conf', dag_id='dag-name', key='connection_name') }} instead of the value test

However, in the Airflow UI, both the tasks have the right rendered value for the key conn_id = test

I have tried all different combinations of using XCOMs and have exhausted all my options. Is there any gotcha in using an XCOM from a task outside a task group for multiple taks in the group? It looks like only the first task (t1) gets the value. One caveat is that t1 also returns an XCOM

I appreciate any suggestion on getting this to work or any workaround for passing XCOM values to tasks in a group

[josh-fell]

Hi @sid-habu! Just curious, have you tried not wrapping the conn_id arg you pass into the operator and sensor as an f-string? Meaning, instead of

conn_id=f'{data_plane_conf["connection_name"]}',

have you tried this:

conn_id=data_plane_conf["connection_name"],

[sid-habu]

@josh-fell Thanks for your reply. No, it didn't work as the rendered value still shows up as

task_instance.xcom_pull(task_ids='get_data_plane_conf', dag_id='dag-name', key='connection_name') }}

Some more data points that can isolate the issue

• The conn_id templating works in t1 but doesn't work in t2
• t2 has another variable app_name that uses rendering to read a value from the xcom output of t1, and that works

I have updated the code snippet to depict the above data points

[Taragolis]

Attributes rendered after the class constructor (__init__ method) and before execute, poke methods

from __future__ import annotations

from functools import cached_property
from typing import TYPE_CHECKING, Sequence

from airflow.sensors.base import BaseSensorOperator

if TYPE_CHECKING:
    from airflow.utils.context import Context


class CustomSensor(BaseSensorOperator):
    template_fields: Sequence[str] = ("conn_id", "app_name")

    def __init__(
        self,
        *,
        conn_id: str,
        app_name: str,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.conn_id = conn_id
        self.app_name = app_name
    
    @cached_property
    def hook(self):
        return CustomHook(connection_name=self.conn_id)
        
    def poke(self, context: Context) -> bool:
        # Do something with your hook, conn_id already templated, avoid to call it in __init__
        print(self.hook.connection_name)
        # Do something with app_name
        print(self.app_name)

[sid-habu]

@Taragolis That worked! Thanks a lot for solving the puzzle